tmq.c 56.1 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18
/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#include "clientInt.h"
#include "clientLog.h"
#include "parser.h"
H
Haojun Liao 已提交
19
#include "tdatablock.h"
20 21 22
#include "tdef.h"
#include "tglobal.h"
#include "tmsgtype.h"
X
Xiaoyu Wang 已提交
23
#include "tqueue.h"
24
#include "tref.h"
L
Liu Jicong 已提交
25 26
#include "ttimer.h"

L
Liu Jicong 已提交
27 28 29 30 31 32 33 34
int32_t tmqAskEp(tmq_t* tmq, bool async);

typedef struct {
  int8_t inited;
  tmr_h  timer;
} SMqMgmt;

static SMqMgmt tmqMgmt = {0};
35

L
Liu Jicong 已提交
36 37 38 39 40 41
typedef struct {
  int8_t  tmqRspType;
  int32_t epoch;
} SMqRspWrapper;

typedef struct {
L
Liu Jicong 已提交
42 43 44
  int8_t      tmqRspType;
  int32_t     epoch;
  SMqAskEpRsp msg;
L
Liu Jicong 已提交
45 46
} SMqAskEpRspWrapper;

L
Liu Jicong 已提交
47
struct tmq_list_t {
L
Liu Jicong 已提交
48
  SArray container;
L
Liu Jicong 已提交
49
};
L
Liu Jicong 已提交
50

L
Liu Jicong 已提交
51
struct tmq_conf_t {
52 53 54 55 56
  char     clientId[256];
  char     groupId[TSDB_CGROUP_LEN];
  int8_t   autoCommit;
  int8_t   resetOffset;
  int8_t   withTbName;
L
Liu Jicong 已提交
57
  int8_t   useSnapshot;
58 59 60 61 62 63
  uint16_t port;
  int32_t  autoCommitInterval;
  char*    ip;
  char*    user;
  char*    pass;
  /*char*          db;*/
64
  tmq_commit_cb* commitCb;
L
Liu Jicong 已提交
65
  void*          commitCbUserParam;
L
Liu Jicong 已提交
66 67 68
};

struct tmq_t {
L
Liu Jicong 已提交
69
  // conf
L
Liu Jicong 已提交
70 71
  char           groupId[TSDB_CGROUP_LEN];
  char           clientId[256];
72
  int8_t         withTbName;
L
Liu Jicong 已提交
73
  int8_t         useSnapshot;
L
Liu Jicong 已提交
74
  int8_t         autoCommit;
L
Liu Jicong 已提交
75
  int32_t        autoCommitInterval;
L
Liu Jicong 已提交
76
  int32_t        resetOffsetCfg;
L
Liu Jicong 已提交
77
  int64_t        consumerId;
L
Liu Jicong 已提交
78 79
  tmq_commit_cb* commitCb;
  void*          commitCbUserParam;
L
Liu Jicong 已提交
80 81 82 83

  // status
  int8_t  status;
  int32_t epoch;
L
Liu Jicong 已提交
84 85
#if 0
  int8_t  epStatus;
L
Liu Jicong 已提交
86
  int32_t epSkipCnt;
L
Liu Jicong 已提交
87
#endif
L
Liu Jicong 已提交
88 89
  int64_t pollCnt;

L
Liu Jicong 已提交
90 91 92 93 94
  // timer
  tmr_h hbTimer;
  tmr_h reportTimer;
  tmr_h commitTimer;

L
Liu Jicong 已提交
95 96 97 98
  // connection
  STscObj* pTscObj;

  // container
L
Liu Jicong 已提交
99
  SArray*     clientTopics;  // SArray<SMqClientTopic>
L
Liu Jicong 已提交
100
  STaosQueue* mqueue;        // queue of rsp
L
Liu Jicong 已提交
101
  STaosQall*  qall;
L
Liu Jicong 已提交
102 103 104 105
  STaosQueue* delayedTask;  // delayed task queue for heartbeat and auto commit

  // ctl
  tsem_t rspSem;
L
Liu Jicong 已提交
106 107
};

X
Xiaoyu Wang 已提交
108 109 110 111 112 113 114 115
enum {
  TMQ_VG_STATUS__IDLE = 0,
  TMQ_VG_STATUS__WAIT,
};

enum {
  TMQ_CONSUMER_STATUS__INIT = 0,
  TMQ_CONSUMER_STATUS__READY,
116
  TMQ_CONSUMER_STATUS__NO_TOPIC,
L
Liu Jicong 已提交
117 118
};

L
Liu Jicong 已提交
119 120 121 122 123 124
enum {
  TMQ_DELAYED_TASK__HB = 1,
  TMQ_DELAYED_TASK__REPORT,
  TMQ_DELAYED_TASK__COMMIT,
};

L
Liu Jicong 已提交
125
typedef struct {
126 127 128
  // statistics
  int64_t pollCnt;
  // offset
129
  int64_t committedOffset;
130
  int64_t currentOffset;
L
Liu Jicong 已提交
131
  // connection info
132
  int32_t vgId;
X
Xiaoyu Wang 已提交
133
  int32_t vgStatus;
L
Liu Jicong 已提交
134
  int32_t vgSkipCnt;
135 136 137
  SEpSet  epSet;
} SMqClientVg;

L
Liu Jicong 已提交
138
typedef struct {
139
  // subscribe info
L
Liu Jicong 已提交
140
  char* topicName;
L
Liu Jicong 已提交
141
  char  db[TSDB_DB_FNAME_LEN];
L
Liu Jicong 已提交
142 143 144

  SArray* vgs;  // SArray<SMqClientVg>

L
Liu Jicong 已提交
145 146
  int8_t         isSchemaAdaptive;
  SSchemaWrapper schema;
147 148
} SMqClientTopic;

L
Liu Jicong 已提交
149 150 151 152 153
typedef struct {
  int8_t          tmqRspType;
  int32_t         epoch;
  SMqClientVg*    vgHandle;
  SMqClientTopic* topicHandle;
L
Liu Jicong 已提交
154 155 156 157
  union {
    SMqDataBlkRsp dataRsp;
    SMqMetaRsp    metaRsp;
  };
L
Liu Jicong 已提交
158 159
} SMqPollRspWrapper;

L
Liu Jicong 已提交
160
typedef struct {
L
Liu Jicong 已提交
161 162 163
  tmq_t*  tmq;
  tsem_t  rspSem;
  int32_t rspErr;
L
Liu Jicong 已提交
164
} SMqSubscribeCbParam;
L
Liu Jicong 已提交
165

L
Liu Jicong 已提交
166
typedef struct {
167
  tmq_t*  tmq;
L
Liu Jicong 已提交
168
  int32_t code;
L
Liu Jicong 已提交
169
  int32_t async;
X
Xiaoyu Wang 已提交
170
  tsem_t  rspSem;
171 172
} SMqAskEpCbParam;

L
Liu Jicong 已提交
173
typedef struct {
L
Liu Jicong 已提交
174 175
  tmq_t*          tmq;
  SMqClientVg*    pVg;
L
Liu Jicong 已提交
176
  SMqClientTopic* pTopic;
L
Liu Jicong 已提交
177
  int32_t         epoch;
L
Liu Jicong 已提交
178
  int32_t         vgId;
L
Liu Jicong 已提交
179
  tsem_t          rspSem;
X
Xiaoyu Wang 已提交
180
} SMqPollCbParam;
181

L
Liu Jicong 已提交
182
typedef struct {
L
Liu Jicong 已提交
183
  tmq_t*         tmq;
L
Liu Jicong 已提交
184 185
  int8_t         async;
  int8_t         automatic;
L
Liu Jicong 已提交
186
  int8_t         freeOffsets;
L
Liu Jicong 已提交
187
  tmq_commit_cb* userCb;
L
Liu Jicong 已提交
188
  tsem_t         rspSem;
L
Liu Jicong 已提交
189
  int32_t        rspErr;
L
Liu Jicong 已提交
190
  SArray*        offsets;
L
Liu Jicong 已提交
191
  void*          userParam;
L
Liu Jicong 已提交
192
} SMqCommitCbParam;
L
Liu Jicong 已提交
193

194 195 196 197 198
typedef struct {
  tmq_t*         tmq;
  int8_t         automatic;
  int8_t         async;
  int8_t         freeOffsets;
L
Liu Jicong 已提交
199 200
  int32_t        waitingRspNum;
  int32_t        totalRspNum;
L
Liu Jicong 已提交
201
  int32_t        rspErr;
202
  tmq_commit_cb* userCb;
L
Liu Jicong 已提交
203 204 205 206
  /*SArray*        successfulOffsets;*/
  /*SArray*        failedOffsets;*/
  void*  userParam;
  tsem_t rspSem;
207 208 209 210 211 212 213
} SMqCommitCbParamSet;

typedef struct {
  SMqCommitCbParamSet* params;
  STqOffset*           pOffset;
} SMqCommitCbParam2;

214
tmq_conf_t* tmq_conf_new() {
wafwerar's avatar
wafwerar 已提交
215
  tmq_conf_t* conf = taosMemoryCalloc(1, sizeof(tmq_conf_t));
216
  conf->withTbName = false;
L
Liu Jicong 已提交
217
  conf->autoCommit = true;
L
Liu Jicong 已提交
218
  conf->autoCommitInterval = 5000;
X
Xiaoyu Wang 已提交
219
  conf->resetOffset = TMQ_CONF__RESET_OFFSET__EARLIEAST;
220 221 222
  return conf;
}

L
Liu Jicong 已提交
223
void tmq_conf_destroy(tmq_conf_t* conf) {
L
Liu Jicong 已提交
224 225 226 227 228 229
  if (conf) {
    if (conf->ip) taosMemoryFree(conf->ip);
    if (conf->user) taosMemoryFree(conf->user);
    if (conf->pass) taosMemoryFree(conf->pass);
    taosMemoryFree(conf);
  }
L
Liu Jicong 已提交
230 231 232
}

tmq_conf_res_t tmq_conf_set(tmq_conf_t* conf, const char* key, const char* value) {
233 234
  if (strcmp(key, "group.id") == 0) {
    strcpy(conf->groupId, value);
L
Liu Jicong 已提交
235
    return TMQ_CONF_OK;
236
  }
L
Liu Jicong 已提交
237

238 239
  if (strcmp(key, "client.id") == 0) {
    strcpy(conf->clientId, value);
L
Liu Jicong 已提交
240 241
    return TMQ_CONF_OK;
  }
L
Liu Jicong 已提交
242

L
Liu Jicong 已提交
243 244
  if (strcmp(key, "enable.auto.commit") == 0) {
    if (strcmp(value, "true") == 0) {
L
Liu Jicong 已提交
245
      conf->autoCommit = true;
L
Liu Jicong 已提交
246 247
      return TMQ_CONF_OK;
    } else if (strcmp(value, "false") == 0) {
L
Liu Jicong 已提交
248
      conf->autoCommit = false;
L
Liu Jicong 已提交
249 250 251 252
      return TMQ_CONF_OK;
    } else {
      return TMQ_CONF_INVALID;
    }
253
  }
L
Liu Jicong 已提交
254

L
Liu Jicong 已提交
255 256 257 258 259
  if (strcmp(key, "auto.commit.interval.ms") == 0) {
    conf->autoCommitInterval = atoi(value);
    return TMQ_CONF_OK;
  }

L
Liu Jicong 已提交
260 261 262 263 264 265 266 267 268 269 270 271 272 273
  if (strcmp(key, "auto.offset.reset") == 0) {
    if (strcmp(value, "none") == 0) {
      conf->resetOffset = TMQ_CONF__RESET_OFFSET__NONE;
      return TMQ_CONF_OK;
    } else if (strcmp(value, "earliest") == 0) {
      conf->resetOffset = TMQ_CONF__RESET_OFFSET__EARLIEAST;
      return TMQ_CONF_OK;
    } else if (strcmp(value, "latest") == 0) {
      conf->resetOffset = TMQ_CONF__RESET_OFFSET__LATEST;
      return TMQ_CONF_OK;
    } else {
      return TMQ_CONF_INVALID;
    }
  }
L
Liu Jicong 已提交
274

275 276
  if (strcmp(key, "msg.with.table.name") == 0) {
    if (strcmp(value, "true") == 0) {
277
      conf->withTbName = true;
L
Liu Jicong 已提交
278
      return TMQ_CONF_OK;
279
    } else if (strcmp(value, "false") == 0) {
280
      conf->withTbName = false;
L
Liu Jicong 已提交
281
      return TMQ_CONF_OK;
282 283 284 285 286
    } else {
      return TMQ_CONF_INVALID;
    }
  }

L
Liu Jicong 已提交
287 288 289 290 291 292 293 294 295 296 297 298
  if (strcmp(key, "experiment.use.snapshot") == 0) {
    if (strcmp(value, "true") == 0) {
      conf->useSnapshot = true;
      return TMQ_CONF_OK;
    } else if (strcmp(value, "false") == 0) {
      conf->useSnapshot = false;
      return TMQ_CONF_OK;
    } else {
      return TMQ_CONF_INVALID;
    }
  }

L
Liu Jicong 已提交
299
  if (strcmp(key, "td.connect.ip") == 0) {
L
Liu Jicong 已提交
300 301 302
    conf->ip = strdup(value);
    return TMQ_CONF_OK;
  }
L
Liu Jicong 已提交
303
  if (strcmp(key, "td.connect.user") == 0) {
L
Liu Jicong 已提交
304 305 306
    conf->user = strdup(value);
    return TMQ_CONF_OK;
  }
L
Liu Jicong 已提交
307
  if (strcmp(key, "td.connect.pass") == 0) {
L
Liu Jicong 已提交
308 309 310
    conf->pass = strdup(value);
    return TMQ_CONF_OK;
  }
L
Liu Jicong 已提交
311
  if (strcmp(key, "td.connect.port") == 0) {
L
Liu Jicong 已提交
312 313 314
    conf->port = atoi(value);
    return TMQ_CONF_OK;
  }
L
Liu Jicong 已提交
315
  if (strcmp(key, "td.connect.db") == 0) {
316
    /*conf->db = strdup(value);*/
L
Liu Jicong 已提交
317 318 319
    return TMQ_CONF_OK;
  }

L
Liu Jicong 已提交
320
  return TMQ_CONF_UNKNOWN;
321 322 323
}

tmq_list_t* tmq_list_new() {
L
Liu Jicong 已提交
324 325
  //
  return (tmq_list_t*)taosArrayInit(0, sizeof(void*));
326 327
}

L
Liu Jicong 已提交
328 329 330
int32_t tmq_list_append(tmq_list_t* list, const char* src) {
  SArray* container = &list->container;
  char*   topic = strdup(src);
L
fix  
Liu Jicong 已提交
331
  if (taosArrayPush(container, &topic) == NULL) return -1;
332 333 334
  return 0;
}

L
Liu Jicong 已提交
335
void tmq_list_destroy(tmq_list_t* list) {
L
Liu Jicong 已提交
336
  SArray* container = &list->container;
L
Liu Jicong 已提交
337
  taosArrayDestroyP(container, taosMemoryFree);
L
Liu Jicong 已提交
338 339
}

L
Liu Jicong 已提交
340 341 342 343 344 345 346 347 348 349
int32_t tmq_list_get_size(const tmq_list_t* list) {
  const SArray* container = &list->container;
  return taosArrayGetSize(container);
}

char** tmq_list_to_c_array(const tmq_list_t* list) {
  const SArray* container = &list->container;
  return container->pData;
}

L
Liu Jicong 已提交
350 351 352 353
static int32_t tmqMakeTopicVgKey(char* dst, const char* topicName, int32_t vg) {
  return sprintf(dst, "%s:%d", topicName, vg);
}

L
Liu Jicong 已提交
354 355
int32_t tmqCommitCb(void* param, const SDataBuf* pMsg, int32_t code) {
  SMqCommitCbParam* pParam = (SMqCommitCbParam*)param;
L
Liu Jicong 已提交
356
  pParam->rspErr = code;
L
Liu Jicong 已提交
357 358
  if (pParam->async) {
    if (pParam->automatic && pParam->tmq->commitCb) {
L
Liu Jicong 已提交
359
      pParam->tmq->commitCb(pParam->tmq, pParam->rspErr, pParam->tmq->commitCbUserParam);
L
Liu Jicong 已提交
360
    } else if (!pParam->automatic && pParam->userCb) {
L
Liu Jicong 已提交
361
      pParam->userCb(pParam->tmq, pParam->rspErr, pParam->userParam);
L
Liu Jicong 已提交
362 363
    }

L
Liu Jicong 已提交
364
    if (pParam->freeOffsets) {
L
Liu Jicong 已提交
365 366 367 368 369 370 371 372 373 374
      taosArrayDestroy(pParam->offsets);
    }

    taosMemoryFree(pParam);
  } else {
    tsem_post(&pParam->rspSem);
  }
  return 0;
}

375 376 377 378
int32_t tmqCommitCb2(void* param, const SDataBuf* pBuf, int32_t code) {
  SMqCommitCbParam2*   pParam = (SMqCommitCbParam2*)param;
  SMqCommitCbParamSet* pParamSet = (SMqCommitCbParamSet*)pParam->params;
  // push into array
L
Liu Jicong 已提交
379
#if 0
380 381 382 383 384
  if (code == 0) {
    taosArrayPush(pParamSet->failedOffsets, &pParam->pOffset);
  } else {
    taosArrayPush(pParamSet->successfulOffsets, &pParam->pOffset);
  }
L
Liu Jicong 已提交
385
#endif
L
Liu Jicong 已提交
386

L
Liu Jicong 已提交
387 388 389
  /*tscDebug("receive offset commit cb of %s on vg %d, offset is %ld", pParam->pOffset->subKey, pParam->->vgId,
   * pOffset->version);*/

390
  // count down waiting rsp
L
Liu Jicong 已提交
391
  int32_t waitingRspNum = atomic_sub_fetch_32(&pParamSet->waitingRspNum, 1);
392 393 394 395 396 397 398
  ASSERT(waitingRspNum >= 0);

  if (waitingRspNum == 0) {
    // if no more waiting rsp
    if (pParamSet->async) {
      // call async cb func
      if (pParamSet->automatic && pParamSet->tmq->commitCb) {
L
Liu Jicong 已提交
399
        pParamSet->tmq->commitCb(pParamSet->tmq, pParamSet->rspErr, pParamSet->tmq->commitCbUserParam);
400 401
      } else if (!pParamSet->automatic && pParamSet->userCb) {
        // sem post
L
Liu Jicong 已提交
402
        pParamSet->userCb(pParamSet->tmq, pParamSet->rspErr, pParamSet->userParam);
403
      }
L
Liu Jicong 已提交
404 405
    } else {
      tsem_post(&pParamSet->rspSem);
406 407
    }

L
Liu Jicong 已提交
408
#if 0
409 410
    taosArrayDestroyP(pParamSet->successfulOffsets, taosMemoryFree);
    taosArrayDestroyP(pParamSet->failedOffsets, taosMemoryFree);
L
Liu Jicong 已提交
411
#endif
412 413 414 415
  }
  return 0;
}

L
Liu Jicong 已提交
416 417
int32_t tmqCommitInner2(tmq_t* tmq, const TAOS_RES* msg, int8_t automatic, int8_t async, tmq_commit_cb* userCb,
                        void* userParam) {
418 419
  int32_t code = -1;

L
Liu Jicong 已提交
420
  if (msg != NULL) {
421 422 423 424 425 426 427 428 429 430 431
    char*   topic;
    int32_t vgId;
    if (TD_RES_TMQ(msg)) {
      SMqRspObj* pRspObj = (SMqRspObj*)msg;
      topic = pRspObj->topic;
      vgId = pRspObj->vgId;
    } else if (TD_RES_TMQ_META(msg)) {
      SMqMetaRspObj* pMetaRspObj = (SMqMetaRspObj*)msg;
      topic = pMetaRspObj->topic;
      vgId = pMetaRspObj->vgId;
    } else {
L
Liu Jicong 已提交
432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449
      return TSDB_CODE_TMQ_INVALID_MSG;
    }

    SMqCommitCbParamSet* pParamSet = taosMemoryCalloc(1, sizeof(SMqCommitCbParamSet));
    if (pParamSet == NULL) {
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      return -1;
    }
    pParamSet->tmq = tmq;
    pParamSet->automatic = automatic;
    pParamSet->async = async;
    pParamSet->freeOffsets = 1;
    pParamSet->userCb = userCb;
    pParamSet->userParam = userParam;
    tsem_init(&pParamSet->rspSem, 0, 0);

    for (int32_t i = 0; i < taosArrayGetSize(tmq->clientTopics); i++) {
      SMqClientTopic* pTopic = taosArrayGet(tmq->clientTopics, i);
450
      if (strcmp(pTopic->topicName, topic) == 0) {
L
Liu Jicong 已提交
451 452
        for (int32_t j = 0; j < taosArrayGetSize(pTopic->vgs); j++) {
          SMqClientVg* pVg = taosArrayGet(pTopic->vgs, j);
453
          if (pVg->vgId == vgId) {
L
Liu Jicong 已提交
454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550
            if (pVg->currentOffset < 0 || pVg->committedOffset == pVg->currentOffset) {
              tscDebug("consumer %ld skip commit for topic %s vg %d, current offset is %ld, committed offset is %ld",
                       tmq->consumerId, pTopic->topicName, pVg->vgId, pVg->currentOffset, pVg->committedOffset);

              return 0;
            }

            STqOffset* pOffset = taosMemoryCalloc(1, sizeof(STqOffset));
            if (pOffset == NULL) {
              terrno = TSDB_CODE_OUT_OF_MEMORY;
              return -1;
            }
            pOffset->type = TMQ_OFFSET__LOG;
            pOffset->version = pVg->currentOffset;

            int32_t groupLen = strlen(tmq->groupId);
            memcpy(pOffset->subKey, tmq->groupId, groupLen);
            pOffset->subKey[groupLen] = TMQ_SEPARATOR;
            strcpy(pOffset->subKey + groupLen + 1, pTopic->topicName);

            int32_t len;
            int32_t code;
            tEncodeSize(tEncodeSTqOffset, pOffset, len, code);
            if (code < 0) {
              ASSERT(0);
            }
            void* buf = taosMemoryCalloc(1, sizeof(SMsgHead) + len);
            ((SMsgHead*)buf)->vgId = htonl(pVg->vgId);

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

            SEncoder encoder;
            tEncoderInit(&encoder, abuf, len);
            tEncodeSTqOffset(&encoder, pOffset);

            // build param
            SMqCommitCbParam2* pParam = taosMemoryCalloc(1, sizeof(SMqCommitCbParam2));
            pParam->params = pParamSet;
            pParam->pOffset = pOffset;

            // build send info
            SMsgSendInfo* pMsgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
            if (pMsgSendInfo == NULL) {
              // TODO
              continue;
            }
            pMsgSendInfo->msgInfo = (SDataBuf){
                .pData = buf,
                .len = sizeof(SMsgHead) + len,
                .handle = NULL,
            };

            tscDebug("consumer %ld commit offset of %s on vg %d, offset is %ld", tmq->consumerId, pOffset->subKey,
                     pVg->vgId, pOffset->version);

            // TODO: put into cb
            pVg->committedOffset = pVg->currentOffset;

            pMsgSendInfo->requestId = generateRequestId();
            pMsgSendInfo->requestObjRefId = 0;
            pMsgSendInfo->param = pParam;
            pMsgSendInfo->fp = tmqCommitCb2;
            pMsgSendInfo->msgType = TDMT_VND_MQ_COMMIT_OFFSET;
            // send msg

            int64_t transporterId = 0;
            asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &pVg->epSet, &transporterId, pMsgSendInfo);
            pParamSet->waitingRspNum++;
            pParamSet->totalRspNum++;
          }
        }
      }
    }
    if (pParamSet->totalRspNum == 0) {
      tsem_destroy(&pParamSet->rspSem);
      taosMemoryFree(pParamSet);
      return 0;
    }

    if (!async) {
      tsem_wait(&pParamSet->rspSem);
      code = pParamSet->rspErr;
      tsem_destroy(&pParamSet->rspSem);
    } else {
      code = 0;
    }

    if (code != 0 && async) {
      if (automatic) {
        tmq->commitCb(tmq, code, tmq->commitCbUserParam);
      } else {
        userCb(tmq, code, userParam);
      }
    }
    return 0;
  }

551 552 553 554 555 556 557 558 559 560 561 562 563 564 565
  SMqCommitCbParamSet* pParamSet = taosMemoryCalloc(1, sizeof(SMqCommitCbParamSet));
  if (pParamSet == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
  }
  pParamSet->tmq = tmq;
  pParamSet->automatic = automatic;
  pParamSet->async = async;
  pParamSet->freeOffsets = 1;
  pParamSet->userCb = userCb;
  pParamSet->userParam = userParam;
  tsem_init(&pParamSet->rspSem, 0, 0);

  for (int32_t i = 0; i < taosArrayGetSize(tmq->clientTopics); i++) {
    SMqClientTopic* pTopic = taosArrayGet(tmq->clientTopics, i);
L
Liu Jicong 已提交
566 567 568 569

    tscDebug("consumer %ld begin commit for topic %s, vgNum %d", tmq->consumerId, pTopic->topicName,
             (int32_t)taosArrayGetSize(pTopic->vgs));

570
    for (int32_t j = 0; j < taosArrayGetSize(pTopic->vgs); j++) {
L
Liu Jicong 已提交
571 572 573 574 575 576 577 578 579
      SMqClientVg* pVg = taosArrayGet(pTopic->vgs, j);

      tscDebug("consumer %ld begin commit for topic %s, vgId %d", tmq->consumerId, pTopic->topicName, pVg->vgId);

      /*if (pVg->currentOffset < 0) {*/
      if (pVg->currentOffset < 0 || pVg->committedOffset == pVg->currentOffset) {
        tscDebug("consumer %ld skip commit for topic %s vg %d, current offset is %ld, committed offset is %ld",
                 tmq->consumerId, pTopic->topicName, pVg->vgId, pVg->currentOffset, pVg->committedOffset);

L
Liu Jicong 已提交
580 581 582
        continue;
      }
      STqOffset* pOffset = taosMemoryCalloc(1, sizeof(STqOffset));
583 584 585 586
      if (pOffset == NULL) {
        terrno = TSDB_CODE_OUT_OF_MEMORY;
        return -1;
      }
L
Liu Jicong 已提交
587 588
      pOffset->type = TMQ_OFFSET__LOG;
      pOffset->version = pVg->currentOffset;
L
Liu Jicong 已提交
589 590 591 592 593 594

      int32_t groupLen = strlen(tmq->groupId);
      memcpy(pOffset->subKey, tmq->groupId, groupLen);
      pOffset->subKey[groupLen] = TMQ_SEPARATOR;
      strcpy(pOffset->subKey + groupLen + 1, pTopic->topicName);

595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622
      int32_t len;
      int32_t code;
      tEncodeSize(tEncodeSTqOffset, pOffset, len, code);
      if (code < 0) {
        ASSERT(0);
      }
      void* buf = taosMemoryCalloc(1, sizeof(SMsgHead) + len);
      ((SMsgHead*)buf)->vgId = htonl(pVg->vgId);

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

      SEncoder encoder;
      tEncoderInit(&encoder, abuf, len);
      tEncodeSTqOffset(&encoder, pOffset);

      // build param
      SMqCommitCbParam2* pParam = taosMemoryCalloc(1, sizeof(SMqCommitCbParam2));
      pParam->params = pParamSet;
      pParam->pOffset = pOffset;

      // build send info
      SMsgSendInfo* pMsgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
      if (pMsgSendInfo == NULL) {
        // TODO
        continue;
      }
      pMsgSendInfo->msgInfo = (SDataBuf){
          .pData = buf,
L
Liu Jicong 已提交
623
          .len = sizeof(SMsgHead) + len,
624 625 626
          .handle = NULL,
      };

L
Liu Jicong 已提交
627 628 629
      tscDebug("consumer %ld commit offset of %s on vg %d, offset is %ld", tmq->consumerId, pOffset->subKey, pVg->vgId,
               pOffset->version);

L
Liu Jicong 已提交
630 631 632
      // TODO: put into cb
      pVg->committedOffset = pVg->currentOffset;

633 634 635 636
      pMsgSendInfo->requestId = generateRequestId();
      pMsgSendInfo->requestObjRefId = 0;
      pMsgSendInfo->param = pParam;
      pMsgSendInfo->fp = tmqCommitCb2;
L
Liu Jicong 已提交
637
      pMsgSendInfo->msgType = TDMT_VND_MQ_COMMIT_OFFSET;
638 639 640
      // send msg

      int64_t transporterId = 0;
L
Liu Jicong 已提交
641
      asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &pVg->epSet, &transporterId, pMsgSendInfo);
L
Liu Jicong 已提交
642 643
      pParamSet->waitingRspNum++;
      pParamSet->totalRspNum++;
644 645 646
    }
  }

L
Liu Jicong 已提交
647 648 649 650 651 652
  if (pParamSet->totalRspNum == 0) {
    tsem_destroy(&pParamSet->rspSem);
    taosMemoryFree(pParamSet);
    return 0;
  }

653 654 655 656 657 658 659 660 661 662
  if (!async) {
    tsem_wait(&pParamSet->rspSem);
    code = pParamSet->rspErr;
    tsem_destroy(&pParamSet->rspSem);
  } else {
    code = 0;
  }

  if (code != 0 && async) {
    if (automatic) {
L
Liu Jicong 已提交
663
      tmq->commitCb(tmq, code, tmq->commitCbUserParam);
664
    } else {
L
Liu Jicong 已提交
665
      userCb(tmq, code, userParam);
666 667 668
    }
  }

L
Liu Jicong 已提交
669
#if 0
670 671 672 673
  if (!async) {
    taosArrayDestroyP(pParamSet->successfulOffsets, taosMemoryFree);
    taosArrayDestroyP(pParamSet->failedOffsets, taosMemoryFree);
  }
L
Liu Jicong 已提交
674
#endif
675 676 677 678

  return 0;
}

L
Liu Jicong 已提交
679 680
#if 0
int32_t tmqCommitInner(tmq_t* tmq, const TAOS_RES* msg, int8_t automatic, int8_t async,
L
Liu Jicong 已提交
681 682 683 684 685 686
                       tmq_commit_cb* userCb, void* userParam) {
  SMqCMCommitOffsetReq req;
  SArray*              pOffsets = NULL;
  void*                buf = NULL;
  SMqCommitCbParam*    pParam = NULL;
  SMsgSendInfo*        sendInfo = NULL;
L
Liu Jicong 已提交
687 688
  int8_t               freeOffsets;
  int32_t              code = -1;
L
Liu Jicong 已提交
689

L
Liu Jicong 已提交
690
  if (msg == NULL) {
L
Liu Jicong 已提交
691
    freeOffsets = 1;
L
Liu Jicong 已提交
692 693 694 695 696 697 698 699 700 701 702 703 704 705
    pOffsets = taosArrayInit(0, sizeof(SMqOffset));
    for (int32_t i = 0; i < taosArrayGetSize(tmq->clientTopics); i++) {
      SMqClientTopic* pTopic = taosArrayGet(tmq->clientTopics, i);
      for (int32_t j = 0; j < taosArrayGetSize(pTopic->vgs); j++) {
        SMqClientVg* pVg = taosArrayGet(pTopic->vgs, j);
        SMqOffset    offset;
        tstrncpy(offset.topicName, pTopic->topicName, TSDB_TOPIC_FNAME_LEN);
        tstrncpy(offset.cgroup, tmq->groupId, TSDB_CGROUP_LEN);
        offset.vgId = pVg->vgId;
        offset.offset = pVg->currentOffset;
        taosArrayPush(pOffsets, &offset);
      }
    }
  } else {
L
Liu Jicong 已提交
706
    freeOffsets = 0;
L
Liu Jicong 已提交
707
    pOffsets = (SArray*)&msg->container;
L
Liu Jicong 已提交
708 709 710 711 712
  }

  req.num = (int32_t)pOffsets->size;
  req.offsets = pOffsets->pData;

L
Liu Jicong 已提交
713 714 715 716
  SEncoder encoder;

  tEncoderInit(&encoder, NULL, 0);
  code = tEncodeSMqCMCommitOffsetReq(&encoder, &req);
L
Liu Jicong 已提交
717 718 719
  if (code < 0) {
    goto END;
  }
L
Liu Jicong 已提交
720
  int32_t tlen = encoder.pos;
L
Liu Jicong 已提交
721 722
  buf = taosMemoryMalloc(tlen);
  if (buf == NULL) {
L
Liu Jicong 已提交
723
    tEncoderClear(&encoder);
L
Liu Jicong 已提交
724 725
    goto END;
  }
L
Liu Jicong 已提交
726 727
  tEncoderClear(&encoder);

L
Liu Jicong 已提交
728 729 730 731 732 733 734 735 736 737 738 739
  tEncoderInit(&encoder, buf, tlen);
  tEncodeSMqCMCommitOffsetReq(&encoder, &req);
  tEncoderClear(&encoder);

  pParam = taosMemoryCalloc(1, sizeof(SMqCommitCbParam));
  if (pParam == NULL) {
    goto END;
  }
  pParam->tmq = tmq;
  pParam->automatic = automatic;
  pParam->async = async;
  pParam->offsets = pOffsets;
L
Liu Jicong 已提交
740
  pParam->freeOffsets = freeOffsets;
L
Liu Jicong 已提交
741 742 743 744
  pParam->userCb = userCb;
  pParam->userParam = userParam;
  if (!async) tsem_init(&pParam->rspSem, 0, 0);

745
  sendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
L
Liu Jicong 已提交
746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768
  if (sendInfo == NULL) goto END;
  sendInfo->msgInfo = (SDataBuf){
      .pData = buf,
      .len = tlen,
      .handle = NULL,
  };

  sendInfo->requestId = generateRequestId();
  sendInfo->requestObjRefId = 0;
  sendInfo->param = pParam;
  sendInfo->fp = tmqCommitCb;
  sendInfo->msgType = TDMT_MND_MQ_COMMIT_OFFSET;

  SEpSet epSet = getEpSet_s(&tmq->pTscObj->pAppInfo->mgmtEp);

  int64_t transporterId = 0;
  asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &epSet, &transporterId, sendInfo);

  if (!async) {
    tsem_wait(&pParam->rspSem);
    code = pParam->rspErr;
    tsem_destroy(&pParam->rspSem);
    taosMemoryFree(pParam);
L
Liu Jicong 已提交
769 770
  } else {
    code = 0;
L
Liu Jicong 已提交
771 772 773 774 775 776 777
  }

  // avoid double free if msg is sent
  buf = NULL;

END:
  if (buf) taosMemoryFree(buf);
L
Liu Jicong 已提交
778 779
  /*if (pParam) taosMemoryFree(pParam);*/
  /*if (sendInfo) taosMemoryFree(sendInfo);*/
L
Liu Jicong 已提交
780 781 782

  if (code != 0 && async) {
    if (automatic) {
L
Liu Jicong 已提交
783
      tmq->commitCb(tmq, code, (tmq_topic_vgroup_list_t*)pOffsets, tmq->commitCbUserParam);
L
Liu Jicong 已提交
784
    } else {
L
Liu Jicong 已提交
785
      userCb(tmq, code, (tmq_topic_vgroup_list_t*)pOffsets, userParam);
L
Liu Jicong 已提交
786 787 788
    }
  }

L
Liu Jicong 已提交
789
  if (!async && freeOffsets) {
L
Liu Jicong 已提交
790 791 792 793
    taosArrayDestroy(pOffsets);
  }
  return code;
}
L
Liu Jicong 已提交
794
#endif
L
Liu Jicong 已提交
795

L
Liu Jicong 已提交
796 797
void tmqAssignDelayedHbTask(void* param, void* tmrId) {
  tmq_t*  tmq = (tmq_t*)param;
798
  int8_t* pTaskType = taosAllocateQitem(sizeof(int8_t), DEF_QITEM);
L
Liu Jicong 已提交
799 800
  *pTaskType = TMQ_DELAYED_TASK__HB;
  taosWriteQitem(tmq->delayedTask, pTaskType);
801
  tsem_post(&tmq->rspSem);
L
Liu Jicong 已提交
802 803 804 805
}

void tmqAssignDelayedCommitTask(void* param, void* tmrId) {
  tmq_t*  tmq = (tmq_t*)param;
806
  int8_t* pTaskType = taosAllocateQitem(sizeof(int8_t), DEF_QITEM);
L
Liu Jicong 已提交
807 808
  *pTaskType = TMQ_DELAYED_TASK__COMMIT;
  taosWriteQitem(tmq->delayedTask, pTaskType);
809
  tsem_post(&tmq->rspSem);
L
Liu Jicong 已提交
810 811 812 813
}

void tmqAssignDelayedReportTask(void* param, void* tmrId) {
  tmq_t*  tmq = (tmq_t*)param;
814
  int8_t* pTaskType = taosAllocateQitem(sizeof(int8_t), DEF_QITEM);
L
Liu Jicong 已提交
815 816
  *pTaskType = TMQ_DELAYED_TASK__REPORT;
  taosWriteQitem(tmq->delayedTask, pTaskType);
817
  tsem_post(&tmq->rspSem);
L
Liu Jicong 已提交
818 819 820 821 822 823 824 825 826 827 828
}

int32_t tmqHandleAllDelayedTask(tmq_t* tmq) {
  STaosQall* qall = taosAllocateQall();
  taosReadAllQitems(tmq->delayedTask, qall);
  while (1) {
    int8_t* pTaskType = NULL;
    taosGetQitem(qall, (void**)&pTaskType);
    if (pTaskType == NULL) break;

    if (*pTaskType == TMQ_DELAYED_TASK__HB) {
L
Liu Jicong 已提交
829
      tmqAskEp(tmq, true);
L
Liu Jicong 已提交
830 831
      taosTmrReset(tmqAssignDelayedHbTask, 1000, tmq, tmqMgmt.timer, &tmq->hbTimer);
    } else if (*pTaskType == TMQ_DELAYED_TASK__COMMIT) {
L
Liu Jicong 已提交
832
      tmqCommitInner2(tmq, NULL, 1, 1, tmq->commitCb, tmq->commitCbUserParam);
L
Liu Jicong 已提交
833 834 835 836 837
      taosTmrReset(tmqAssignDelayedCommitTask, tmq->autoCommitInterval, tmq, tmqMgmt.timer, &tmq->commitTimer);
    } else if (*pTaskType == TMQ_DELAYED_TASK__REPORT) {
    } else {
      ASSERT(0);
    }
L
Liu Jicong 已提交
838
    taosFreeQitem(pTaskType);
L
Liu Jicong 已提交
839 840 841 842 843
  }
  taosFreeQall(qall);
  return 0;
}

L
Liu Jicong 已提交
844
void tmqClearUnhandleMsg(tmq_t* tmq) {
L
Liu Jicong 已提交
845
  SMqRspWrapper* msg = NULL;
L
Liu Jicong 已提交
846 847 848 849 850 851 852 853
  while (1) {
    taosGetQitem(tmq->qall, (void**)&msg);
    if (msg)
      taosFreeQitem(msg);
    else
      break;
  }

L
fix  
Liu Jicong 已提交
854
  msg = NULL;
L
Liu Jicong 已提交
855 856 857 858 859 860 861 862 863 864
  taosReadAllQitems(tmq->mqueue, tmq->qall);
  while (1) {
    taosGetQitem(tmq->qall, (void**)&msg);
    if (msg)
      taosFreeQitem(msg);
    else
      break;
  }
}

L
Liu Jicong 已提交
865 866 867
int32_t tmqSubscribeCb(void* param, const SDataBuf* pMsg, int32_t code) {
  SMqSubscribeCbParam* pParam = (SMqSubscribeCbParam*)param;
  pParam->rspErr = code;
L
Liu Jicong 已提交
868
  /*tmq_t* tmq = pParam->tmq;*/
L
Liu Jicong 已提交
869 870 871
  tsem_post(&pParam->rspSem);
  return 0;
}
872

L
Liu Jicong 已提交
873
int32_t tmq_subscription(tmq_t* tmq, tmq_list_t** topics) {
X
Xiaoyu Wang 已提交
874 875 876 877
  if (*topics == NULL) {
    *topics = tmq_list_new();
  }
  for (int i = 0; i < taosArrayGetSize(tmq->clientTopics); i++) {
L
Liu Jicong 已提交
878
    SMqClientTopic* topic = taosArrayGet(tmq->clientTopics, i);
L
Liu Jicong 已提交
879
    tmq_list_append(*topics, strchr(topic->topicName, '.') + 1);
X
Xiaoyu Wang 已提交
880
  }
L
Liu Jicong 已提交
881
  return 0;
X
Xiaoyu Wang 已提交
882 883
}

L
Liu Jicong 已提交
884 885 886
int32_t tmq_unsubscribe(tmq_t* tmq) {
  tmq_list_t* lst = tmq_list_new();
  int32_t     rsp = tmq_subscribe(tmq, lst);
L
Liu Jicong 已提交
887 888
  tmq_list_destroy(lst);
  return rsp;
X
Xiaoyu Wang 已提交
889 890
}

L
Liu Jicong 已提交
891
#if 0
892
tmq_t* tmq_consumer_new(void* conn, tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
wafwerar's avatar
wafwerar 已提交
893
  tmq_t* pTmq = taosMemoryCalloc(sizeof(tmq_t), 1);
894 895 896 897 898 899
  if (pTmq == NULL) {
    return NULL;
  }
  pTmq->pTscObj = (STscObj*)conn;
  pTmq->status = 0;
  pTmq->pollCnt = 0;
L
Liu Jicong 已提交
900
  pTmq->epoch = 0;
L
fix txn  
Liu Jicong 已提交
901
  pTmq->epStatus = 0;
L
temp  
Liu Jicong 已提交
902
  pTmq->epSkipCnt = 0;
L
Liu Jicong 已提交
903
  // set conf
904 905
  strcpy(pTmq->clientId, conf->clientId);
  strcpy(pTmq->groupId, conf->groupId);
L
Liu Jicong 已提交
906
  pTmq->autoCommit = conf->autoCommit;
907
  pTmq->commit_cb = conf->commit_cb;
L
Liu Jicong 已提交
908
  pTmq->resetOffsetCfg = conf->resetOffset;
L
Liu Jicong 已提交
909

L
Liu Jicong 已提交
910 911 912 913 914 915 916 917 918 919
  pTmq->consumerId = generateRequestId() & (((uint64_t)-1) >> 1);
  pTmq->clientTopics = taosArrayInit(0, sizeof(SMqClientTopic));
  if (pTmq->clientTopics == NULL) {
    taosMemoryFree(pTmq);
    return NULL;
  }

  pTmq->mqueue = taosOpenQueue();
  pTmq->qall = taosAllocateQall();

L
Liu Jicong 已提交
920 921
  tsem_init(&pTmq->rspSem, 0, 0);

L
Liu Jicong 已提交
922 923
  return pTmq;
}
L
Liu Jicong 已提交
924
#endif
L
Liu Jicong 已提交
925

L
Liu Jicong 已提交
926
tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
L
Liu Jicong 已提交
927 928 929 930 931 932 933 934 935 936 937
  // init timer
  int8_t inited = atomic_val_compare_exchange_8(&tmqMgmt.inited, 0, 1);
  if (inited == 0) {
    tmqMgmt.timer = taosTmrInit(1000, 100, 360000, "TMQ");
    if (tmqMgmt.timer == NULL) {
      atomic_store_8(&tmqMgmt.inited, 0);
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      return NULL;
    }
  }

L
Liu Jicong 已提交
938 939 940 941
  tmq_t* pTmq = taosMemoryCalloc(1, sizeof(tmq_t));
  if (pTmq == NULL) {
    return NULL;
  }
L
Liu Jicong 已提交
942

L
Liu Jicong 已提交
943 944 945 946 947
  const char* user = conf->user == NULL ? TSDB_DEFAULT_USER : conf->user;
  const char* pass = conf->pass == NULL ? TSDB_DEFAULT_PASS : conf->pass;

  ASSERT(user);
  ASSERT(pass);
L
Liu Jicong 已提交
948
  ASSERT(conf->groupId[0]);
L
Liu Jicong 已提交
949

L
Liu Jicong 已提交
950 951 952 953 954 955 956 957
  pTmq->clientTopics = taosArrayInit(0, sizeof(SMqClientTopic));
  pTmq->mqueue = taosOpenQueue();
  pTmq->qall = taosAllocateQall();
  pTmq->delayedTask = taosOpenQueue();

  if (pTmq->clientTopics == NULL || pTmq->mqueue == NULL || pTmq->qall == NULL || pTmq->delayedTask == NULL) {
    goto FAIL;
  }
L
Liu Jicong 已提交
958

L
Liu Jicong 已提交
959 960
  // init status
  pTmq->status = TMQ_CONSUMER_STATUS__INIT;
L
Liu Jicong 已提交
961 962
  pTmq->pollCnt = 0;
  pTmq->epoch = 0;
L
Liu Jicong 已提交
963 964
  /*pTmq->epStatus = 0;*/
  /*pTmq->epSkipCnt = 0;*/
L
Liu Jicong 已提交
965

L
Liu Jicong 已提交
966 967 968
  // set conf
  strcpy(pTmq->clientId, conf->clientId);
  strcpy(pTmq->groupId, conf->groupId);
969
  pTmq->withTbName = conf->withTbName;
L
Liu Jicong 已提交
970
  pTmq->useSnapshot = conf->useSnapshot;
L
Liu Jicong 已提交
971
  pTmq->autoCommit = conf->autoCommit;
L
Liu Jicong 已提交
972
  pTmq->autoCommitInterval = conf->autoCommitInterval;
L
Liu Jicong 已提交
973 974
  pTmq->commitCb = conf->commitCb;
  pTmq->commitCbUserParam = conf->commitCbUserParam;
L
Liu Jicong 已提交
975 976
  pTmq->resetOffsetCfg = conf->resetOffset;

L
Liu Jicong 已提交
977
  // assign consumerId
L
Liu Jicong 已提交
978
  pTmq->consumerId = tGenIdPI64();
X
Xiaoyu Wang 已提交
979

L
Liu Jicong 已提交
980 981 982 983
  // init semaphore
  if (tsem_init(&pTmq->rspSem, 0, 0) != 0) {
    goto FAIL;
  }
L
Liu Jicong 已提交
984

L
Liu Jicong 已提交
985 986 987 988 989 990
  // init connection
  pTmq->pTscObj = taos_connect_internal(conf->ip, user, pass, NULL, NULL, conf->port, CONN_TYPE__TMQ);
  if (pTmq->pTscObj == NULL) {
    tsem_destroy(&pTmq->rspSem);
    goto FAIL;
  }
L
Liu Jicong 已提交
991

992
  return pTmq;
L
Liu Jicong 已提交
993 994 995 996 997 998 999 1000

FAIL:
  if (pTmq->clientTopics) taosArrayDestroy(pTmq->clientTopics);
  if (pTmq->mqueue) taosCloseQueue(pTmq->mqueue);
  if (pTmq->delayedTask) taosCloseQueue(pTmq->delayedTask);
  if (pTmq->qall) taosFreeQall(pTmq->qall);
  taosMemoryFree(pTmq);
  return NULL;
1001 1002
}

L
Liu Jicong 已提交
1003 1004
#if 0
int32_t tmq_commit(tmq_t* tmq, const tmq_topic_vgroup_list_t* offsets, int32_t async) {
L
Liu Jicong 已提交
1005
  return tmqCommitInner2(tmq, offsets, 0, async, tmq->commitCb, tmq->commitCbUserParam);
L
Liu Jicong 已提交
1006
}
L
Liu Jicong 已提交
1007
#endif
L
Liu Jicong 已提交
1008

L
Liu Jicong 已提交
1009
int32_t tmq_subscribe(tmq_t* tmq, const tmq_list_t* topic_list) {
L
Liu Jicong 已提交
1010 1011 1012 1013 1014
  const SArray*   container = &topic_list->container;
  int32_t         sz = taosArrayGetSize(container);
  void*           buf = NULL;
  SCMSubscribeReq req = {0};
  int32_t         code = -1;
1015 1016

  req.consumerId = tmq->consumerId;
L
Liu Jicong 已提交
1017
  tstrncpy(req.cgroup, tmq->groupId, TSDB_CGROUP_LEN);
1018
  req.topicNames = taosArrayInit(sz, sizeof(void*));
L
Liu Jicong 已提交
1019
  if (req.topicNames == NULL) goto FAIL;
1020

L
Liu Jicong 已提交
1021 1022
  for (int32_t i = 0; i < sz; i++) {
    char* topic = taosArrayGetP(container, i);
1023 1024

    SName name = {0};
L
Liu Jicong 已提交
1025
    tNameSetDbName(&name, tmq->pTscObj->acctId, topic, strlen(topic));
1026

L
Liu Jicong 已提交
1027 1028 1029
    char* topicFName = taosMemoryCalloc(1, TSDB_TOPIC_FNAME_LEN);
    if (topicFName == NULL) {
      goto FAIL;
1030
    }
L
Liu Jicong 已提交
1031
    tNameExtractFullName(&name, topicFName);
1032

L
Liu Jicong 已提交
1033 1034 1035
    tscDebug("subscribe topic: %s", topicFName);

    taosArrayPush(req.topicNames, &topicFName);
1036 1037
  }

L
Liu Jicong 已提交
1038 1039 1040 1041
  int32_t tlen = tSerializeSCMSubscribeReq(NULL, &req);
  buf = taosMemoryMalloc(tlen);
  if (buf == NULL) goto FAIL;

1042 1043 1044
  void* abuf = buf;
  tSerializeSCMSubscribeReq(&abuf, &req);

1045
  SMsgSendInfo* sendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
L
Liu Jicong 已提交
1046
  if (sendInfo == NULL) goto FAIL;
1047

X
Xiaoyu Wang 已提交
1048
  SMqSubscribeCbParam param = {
L
Liu Jicong 已提交
1049
      .rspErr = 0,
X
Xiaoyu Wang 已提交
1050 1051
      .tmq = tmq,
  };
L
Liu Jicong 已提交
1052

L
Liu Jicong 已提交
1053 1054 1055
  if (tsem_init(&param.rspSem, 0, 0) != 0) goto FAIL;

  sendInfo->msgInfo = (SDataBuf){
X
Xiaoyu Wang 已提交
1056 1057 1058 1059
      .pData = buf,
      .len = tlen,
      .handle = NULL,
  };
1060

L
Liu Jicong 已提交
1061 1062
  sendInfo->requestId = generateRequestId();
  sendInfo->requestObjRefId = 0;
L
Liu Jicong 已提交
1063 1064
  sendInfo->param = &param;
  sendInfo->fp = tmqSubscribeCb;
L
Liu Jicong 已提交
1065 1066
  sendInfo->msgType = TDMT_MND_SUBSCRIBE;

1067 1068 1069 1070 1071
  SEpSet epSet = getEpSet_s(&tmq->pTscObj->pAppInfo->mgmtEp);

  int64_t transporterId = 0;
  asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &epSet, &transporterId, sendInfo);

L
Liu Jicong 已提交
1072 1073 1074
  // avoid double free if msg is sent
  buf = NULL;

L
Liu Jicong 已提交
1075 1076
  tsem_wait(&param.rspSem);
  tsem_destroy(&param.rspSem);
1077

L
Liu Jicong 已提交
1078 1079 1080
  code = param.rspErr;
  if (code != 0) goto FAIL;

L
Liu Jicong 已提交
1081
  while (TSDB_CODE_MND_CONSUMER_NOT_READY == tmqAskEp(tmq, false)) {
L
fix  
Liu Jicong 已提交
1082
    tscDebug("consumer not ready, retry");
L
Liu Jicong 已提交
1083 1084
    taosMsleep(500);
  }
1085

L
Liu Jicong 已提交
1086
  // init hb timer
1087 1088 1089
  if (tmq->hbTimer == NULL) {
    tmq->hbTimer = taosTmrStart(tmqAssignDelayedHbTask, 1000, tmq, tmqMgmt.timer);
  }
L
Liu Jicong 已提交
1090 1091

  // init auto commit timer
1092
  if (tmq->autoCommit && tmq->commitTimer == NULL) {
L
Liu Jicong 已提交
1093 1094 1095
    tmq->commitTimer = taosTmrStart(tmqAssignDelayedCommitTask, tmq->autoCommitInterval, tmq, tmqMgmt.timer);
  }

L
Liu Jicong 已提交
1096 1097 1098
  code = 0;
FAIL:
  if (req.topicNames != NULL) taosArrayDestroyP(req.topicNames, taosMemoryFree);
L
Liu Jicong 已提交
1099
  if (code != 0 && buf) {
L
Liu Jicong 已提交
1100 1101 1102
    taosMemoryFree(buf);
  }
  return code;
1103 1104
}

L
Liu Jicong 已提交
1105
void tmq_conf_set_auto_commit_cb(tmq_conf_t* conf, tmq_commit_cb* cb, void* param) {
L
Liu Jicong 已提交
1106
  //
1107
  conf->commitCb = cb;
L
Liu Jicong 已提交
1108
  conf->commitCbUserParam = param;
L
Liu Jicong 已提交
1109
}
1110

L
Liu Jicong 已提交
1111
#if 0
L
Liu Jicong 已提交
1112 1113
int32_t tmqGetSkipLogNum(tmq_message_t* tmq_message) {
  if (tmq_message == NULL) return 0;
L
Liu Jicong 已提交
1114
  SMqPollRsp* pRsp = &tmq_message->msg;
L
Liu Jicong 已提交
1115 1116
  return pRsp->skipLogNum;
}
L
Liu Jicong 已提交
1117
#endif
L
Liu Jicong 已提交
1118 1119

int32_t tmqPollCb(void* param, const SDataBuf* pMsg, int32_t code) {
X
Xiaoyu Wang 已提交
1120 1121
  SMqPollCbParam* pParam = (SMqPollCbParam*)param;
  SMqClientVg*    pVg = pParam->pVg;
L
Liu Jicong 已提交
1122
  SMqClientTopic* pTopic = pParam->pTopic;
X
Xiaoyu Wang 已提交
1123
  tmq_t*          tmq = pParam->tmq;
L
Liu Jicong 已提交
1124 1125 1126
  int32_t         vgId = pParam->vgId;
  int32_t         epoch = pParam->epoch;
  taosMemoryFree(pParam);
L
Liu Jicong 已提交
1127
  if (code != 0) {
L
Liu Jicong 已提交
1128 1129
    tscWarn("msg discard from vg %d, epoch %d, code:%x", vgId, epoch, code);
    if (pMsg->pData) taosMemoryFree(pMsg->pData);
L
Liu Jicong 已提交
1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142
    if (code == TSDB_CODE_TQ_NO_COMMITTED_OFFSET) {
      SMqPollRspWrapper* pRspWrapper = taosAllocateQitem(sizeof(SMqPollRspWrapper), DEF_QITEM);
      if (pRspWrapper == NULL) {
        taosMemoryFree(pMsg->pData);
        tscWarn("msg discard from vg %d, epoch %d since out of memory", vgId, epoch);
        goto CREATE_MSG_FAIL;
      }
      pRspWrapper->tmqRspType = TMQ_MSG_TYPE__END_RSP;
      /*pRspWrapper->vgHandle = pVg;*/
      /*pRspWrapper->topicHandle = pTopic;*/
      taosWriteQitem(tmq->mqueue, pRspWrapper);
      tsem_post(&tmq->rspSem);
    }
L
fix txn  
Liu Jicong 已提交
1143
    goto CREATE_MSG_FAIL;
L
Liu Jicong 已提交
1144 1145
  }

X
Xiaoyu Wang 已提交
1146 1147 1148
  int32_t msgEpoch = ((SMqRspHead*)pMsg->pData)->epoch;
  int32_t tmqEpoch = atomic_load_32(&tmq->epoch);
  if (msgEpoch < tmqEpoch) {
L
Liu Jicong 已提交
1149
    // do not write into queue since updating epoch reset
L
Liu Jicong 已提交
1150
    tscWarn("msg discard from vg %d since from earlier epoch, rsp epoch %d, current epoch %d", vgId, msgEpoch,
L
Liu Jicong 已提交
1151
            tmqEpoch);
1152
    tsem_post(&tmq->rspSem);
L
Liu Jicong 已提交
1153
    taosMemoryFree(pMsg->pData);
X
Xiaoyu Wang 已提交
1154 1155 1156 1157
    return 0;
  }

  if (msgEpoch != tmqEpoch) {
L
Liu Jicong 已提交
1158
    tscWarn("mismatch rsp from vg %d, epoch %d, current epoch %d", vgId, msgEpoch, tmqEpoch);
X
Xiaoyu Wang 已提交
1159 1160
  }

L
Liu Jicong 已提交
1161 1162 1163
  // handle meta rsp
  int8_t rspType = ((SMqRspHead*)pMsg->pData)->mqMsgType;

1164
  SMqPollRspWrapper* pRspWrapper = taosAllocateQitem(sizeof(SMqPollRspWrapper), DEF_QITEM);
L
Liu Jicong 已提交
1165
  if (pRspWrapper == NULL) {
L
Liu Jicong 已提交
1166 1167
    taosMemoryFree(pMsg->pData);
    tscWarn("msg discard from vg %d, epoch %d since out of memory", vgId, epoch);
L
fix txn  
Liu Jicong 已提交
1168
    goto CREATE_MSG_FAIL;
L
Liu Jicong 已提交
1169
  }
L
Liu Jicong 已提交
1170

L
Liu Jicong 已提交
1171
  pRspWrapper->tmqRspType = rspType;
L
Liu Jicong 已提交
1172 1173
  pRspWrapper->vgHandle = pVg;
  pRspWrapper->topicHandle = pTopic;
L
Liu Jicong 已提交
1174

L
Liu Jicong 已提交
1175
  if (rspType == TMQ_MSG_TYPE__POLL_RSP) {
L
Liu Jicong 已提交
1176
    memcpy(&pRspWrapper->dataRsp, pMsg->pData, sizeof(SMqRspHead));
L
Liu Jicong 已提交
1177 1178 1179
    tDecodeSMqDataBlkRsp(POINTER_SHIFT(pMsg->pData, sizeof(SMqRspHead)), &pRspWrapper->dataRsp);
  } else {
    ASSERT(rspType == TMQ_MSG_TYPE__POLL_META_RSP);
L
Liu Jicong 已提交
1180
    memcpy(&pRspWrapper->metaRsp, pMsg->pData, sizeof(SMqRspHead));
L
Liu Jicong 已提交
1181 1182
    tDecodeSMqMetaRsp(POINTER_SHIFT(pMsg->pData, sizeof(SMqRspHead)), &pRspWrapper->metaRsp);
  }
L
Liu Jicong 已提交
1183

L
Liu Jicong 已提交
1184
  taosMemoryFree(pMsg->pData);
L
Liu Jicong 已提交
1185

L
Liu Jicong 已提交
1186 1187
  tscDebug("consumer %ld recv poll: vg %d, req offset %ld, rsp offset %ld, type %d", tmq->consumerId, pVg->vgId,
           pRspWrapper->dataRsp.reqOffset, pRspWrapper->dataRsp.rspOffset, rspType);
L
fix  
Liu Jicong 已提交
1188

L
Liu Jicong 已提交
1189
  taosWriteQitem(tmq->mqueue, pRspWrapper);
1190
  tsem_post(&tmq->rspSem);
L
Liu Jicong 已提交
1191

L
Liu Jicong 已提交
1192
  return 0;
L
fix txn  
Liu Jicong 已提交
1193
CREATE_MSG_FAIL:
L
Liu Jicong 已提交
1194
  if (epoch == tmq->epoch) {
L
Liu Jicong 已提交
1195 1196
    atomic_store_32(&pVg->vgStatus, TMQ_VG_STATUS__IDLE);
  }
1197
  tsem_post(&tmq->rspSem);
L
Liu Jicong 已提交
1198
  return -1;
1199 1200
}

1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220 1221 1222 1223 1224 1225 1226 1227 1228
bool tmqUpdateEp2(tmq_t* tmq, int32_t epoch, SMqAskEpRsp* pRsp) {
  bool set = false;

  int32_t topicNumGet = taosArrayGetSize(pRsp->topics);
  char    vgKey[TSDB_TOPIC_FNAME_LEN + 22];
  tscDebug("consumer %ld update ep epoch %d to epoch %d, topic num: %d", tmq->consumerId, tmq->epoch, epoch,
           topicNumGet);

  SArray* newTopics = taosArrayInit(topicNumGet, sizeof(SMqClientTopic));
  if (newTopics == NULL) {
    return false;
  }

  SHashObj* pHash = taosHashInit(64, MurmurHash3_32, false, HASH_NO_LOCK);
  if (pHash == NULL) {
    taosArrayDestroy(newTopics);
    return false;
  }
  int32_t topicNumCur = taosArrayGetSize(tmq->clientTopics);
  for (int32_t i = 0; i < topicNumCur; i++) {
    // find old topic
    SMqClientTopic* pTopicCur = taosArrayGet(tmq->clientTopics, i);
    if (pTopicCur->vgs) {
      int32_t vgNumCur = taosArrayGetSize(pTopicCur->vgs);
      tscDebug("consumer %ld new vg num: %d", tmq->consumerId, vgNumCur);
      for (int32_t j = 0; j < vgNumCur; j++) {
        SMqClientVg* pVgCur = taosArrayGet(pTopicCur->vgs, j);
        sprintf(vgKey, "%s:%d", pTopicCur->topicName, pVgCur->vgId);
L
Liu Jicong 已提交
1229 1230
        tscDebug("consumer %ld epoch %d vg %d vgKey is %s, offset is %ld", tmq->consumerId, epoch, pVgCur->vgId, vgKey,
                 pVgCur->currentOffset);
1231 1232 1233 1234 1235 1236 1237 1238 1239 1240 1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255
        taosHashPut(pHash, vgKey, strlen(vgKey), &pVgCur->currentOffset, sizeof(int64_t));
      }
    }
  }

  for (int32_t i = 0; i < topicNumGet; i++) {
    SMqClientTopic topic = {0};
    SMqSubTopicEp* pTopicEp = taosArrayGet(pRsp->topics, i);
    topic.schema = pTopicEp->schema;
    topic.topicName = strdup(pTopicEp->topic);
    tstrncpy(topic.db, pTopicEp->db, TSDB_DB_FNAME_LEN);

    tscDebug("consumer %ld update topic: %s", tmq->consumerId, topic.topicName);

    int32_t vgNumGet = taosArrayGetSize(pTopicEp->vgs);
    topic.vgs = taosArrayInit(vgNumGet, sizeof(SMqClientVg));
    for (int32_t j = 0; j < vgNumGet; j++) {
      SMqSubVgEp* pVgEp = taosArrayGet(pTopicEp->vgs, j);
      sprintf(vgKey, "%s:%d", topic.topicName, pVgEp->vgId);
      int64_t* pOffset = taosHashGet(pHash, vgKey, strlen(vgKey));
      int64_t  offset = tmq->resetOffsetCfg;
      if (pOffset != NULL) {
        offset = *pOffset;
      }

L
Liu Jicong 已提交
1256 1257
      tscDebug("consumer %ld(epoch %d) offset of vg %d updated to %ld, vgKey is %s", tmq->consumerId, epoch,
               pVgEp->vgId, offset, vgKey);
1258 1259 1260 1261 1262 1263 1264 1265 1266 1267 1268 1269 1270 1271 1272 1273 1274 1275 1276 1277 1278 1279 1280 1281 1282 1283
      SMqClientVg clientVg = {
          .pollCnt = 0,
          .currentOffset = offset,
          .vgId = pVgEp->vgId,
          .epSet = pVgEp->epSet,
          .vgStatus = TMQ_VG_STATUS__IDLE,
          .vgSkipCnt = 0,
      };
      taosArrayPush(topic.vgs, &clientVg);
      set = true;
    }
    taosArrayPush(newTopics, &topic);
  }
  if (tmq->clientTopics) taosArrayDestroy(tmq->clientTopics);
  taosHashCleanup(pHash);
  tmq->clientTopics = newTopics;

  if (taosArrayGetSize(tmq->clientTopics) == 0)
    atomic_store_8(&tmq->status, TMQ_CONSUMER_STATUS__NO_TOPIC);
  else
    atomic_store_8(&tmq->status, TMQ_CONSUMER_STATUS__READY);

  atomic_store_32(&tmq->epoch, epoch);
  return set;
}

L
Liu Jicong 已提交
1284
#if 1
L
Liu Jicong 已提交
1285
bool tmqUpdateEp(tmq_t* tmq, int32_t epoch, SMqAskEpRsp* pRsp) {
L
Liu Jicong 已提交
1286
  /*printf("call update ep %d\n", epoch);*/
X
Xiaoyu Wang 已提交
1287
  bool    set = false;
L
Liu Jicong 已提交
1288 1289
  int32_t topicNumGet = taosArrayGetSize(pRsp->topics);
  char    vgKey[TSDB_TOPIC_FNAME_LEN + 22];
L
Liu Jicong 已提交
1290 1291
  tscDebug("consumer %ld update ep epoch %d to epoch %d, topic num: %d", tmq->consumerId, tmq->epoch, epoch,
           topicNumGet);
L
Liu Jicong 已提交
1292 1293 1294 1295 1296 1297 1298 1299 1300 1301 1302 1303
  SArray* newTopics = taosArrayInit(topicNumGet, sizeof(SMqClientTopic));
  if (newTopics == NULL) {
    return false;
  }
  SHashObj* pHash = taosHashInit(64, MurmurHash3_32, false, HASH_NO_LOCK);
  if (pHash == NULL) {
    taosArrayDestroy(newTopics);
    return false;
  }

  // find topic, build hash
  for (int32_t i = 0; i < topicNumGet; i++) {
X
Xiaoyu Wang 已提交
1304 1305
    SMqClientTopic topic = {0};
    SMqSubTopicEp* pTopicEp = taosArrayGet(pRsp->topics, i);
L
Liu Jicong 已提交
1306
    topic.schema = pTopicEp->schema;
L
Liu Jicong 已提交
1307
    taosHashClear(pHash);
X
Xiaoyu Wang 已提交
1308
    topic.topicName = strdup(pTopicEp->topic);
L
Liu Jicong 已提交
1309
    tstrncpy(topic.db, pTopicEp->db, TSDB_DB_FNAME_LEN);
L
Liu Jicong 已提交
1310

L
Liu Jicong 已提交
1311
    tscDebug("consumer %ld update topic: %s", tmq->consumerId, topic.topicName);
L
Liu Jicong 已提交
1312 1313 1314 1315 1316 1317
    int32_t topicNumCur = taosArrayGetSize(tmq->clientTopics);
    for (int32_t j = 0; j < topicNumCur; j++) {
      // find old topic
      SMqClientTopic* pTopicCur = taosArrayGet(tmq->clientTopics, j);
      if (pTopicCur->vgs && strcmp(pTopicCur->topicName, pTopicEp->topic) == 0) {
        int32_t vgNumCur = taosArrayGetSize(pTopicCur->vgs);
L
Liu Jicong 已提交
1318
        tscDebug("consumer %ld new vg num: %d", tmq->consumerId, vgNumCur);
L
Liu Jicong 已提交
1319 1320 1321 1322
        if (vgNumCur == 0) break;
        for (int32_t k = 0; k < vgNumCur; k++) {
          SMqClientVg* pVgCur = taosArrayGet(pTopicCur->vgs, k);
          sprintf(vgKey, "%s:%d", topic.topicName, pVgCur->vgId);
L
Liu Jicong 已提交
1323
          tscDebug("consumer %ld epoch %d vg %d build %s", tmq->consumerId, epoch, pVgCur->vgId, vgKey);
L
Liu Jicong 已提交
1324 1325 1326 1327 1328 1329 1330 1331 1332
          taosHashPut(pHash, vgKey, strlen(vgKey), &pVgCur->currentOffset, sizeof(int64_t));
        }
        break;
      }
    }

    int32_t vgNumGet = taosArrayGetSize(pTopicEp->vgs);
    topic.vgs = taosArrayInit(vgNumGet, sizeof(SMqClientVg));
    for (int32_t j = 0; j < vgNumGet; j++) {
X
Xiaoyu Wang 已提交
1333
      SMqSubVgEp* pVgEp = taosArrayGet(pTopicEp->vgs, j);
L
Liu Jicong 已提交
1334 1335 1336
      sprintf(vgKey, "%s:%d", topic.topicName, pVgEp->vgId);
      int64_t* pOffset = taosHashGet(pHash, vgKey, strlen(vgKey));
      int64_t  offset = pVgEp->offset;
1337
      tscDebug("consumer %ld(epoch %d) original offset of vg %d is %ld", tmq->consumerId, epoch, pVgEp->vgId, offset);
L
Liu Jicong 已提交
1338 1339
      if (pOffset != NULL) {
        offset = *pOffset;
1340 1341
        tscDebug("consumer %ld(epoch %d) receive offset of vg %d, full key is %s", tmq->consumerId, epoch, pVgEp->vgId,
                 vgKey);
L
Liu Jicong 已提交
1342
      }
1343
      tscDebug("consumer %ld(epoch %d) offset of vg %d updated to %ld", tmq->consumerId, epoch, pVgEp->vgId, offset);
X
Xiaoyu Wang 已提交
1344 1345
      SMqClientVg clientVg = {
          .pollCnt = 0,
L
Liu Jicong 已提交
1346
          .currentOffset = offset,
X
Xiaoyu Wang 已提交
1347 1348 1349
          .vgId = pVgEp->vgId,
          .epSet = pVgEp->epSet,
          .vgStatus = TMQ_VG_STATUS__IDLE,
L
Liu Jicong 已提交
1350
          .vgSkipCnt = 0,
X
Xiaoyu Wang 已提交
1351 1352 1353 1354
      };
      taosArrayPush(topic.vgs, &clientVg);
      set = true;
    }
L
Liu Jicong 已提交
1355
    taosArrayPush(newTopics, &topic);
X
Xiaoyu Wang 已提交
1356
  }
L
Liu Jicong 已提交
1357
  if (tmq->clientTopics) taosArrayDestroy(tmq->clientTopics);
L
Liu Jicong 已提交
1358
  taosHashCleanup(pHash);
L
Liu Jicong 已提交
1359
  tmq->clientTopics = newTopics;
1360

1361 1362 1363 1364
  if (taosArrayGetSize(tmq->clientTopics) == 0)
    atomic_store_8(&tmq->status, TMQ_CONSUMER_STATUS__NO_TOPIC);
  else
    atomic_store_8(&tmq->status, TMQ_CONSUMER_STATUS__READY);
1365

X
Xiaoyu Wang 已提交
1366 1367 1368
  atomic_store_32(&tmq->epoch, epoch);
  return set;
}
L
Liu Jicong 已提交
1369
#endif
X
Xiaoyu Wang 已提交
1370

L
Liu Jicong 已提交
1371
int32_t tmqAskEpCb(void* param, const SDataBuf* pMsg, int32_t code) {
1372
  SMqAskEpCbParam* pParam = (SMqAskEpCbParam*)param;
L
Liu Jicong 已提交
1373
  tmq_t*           tmq = pParam->tmq;
L
Liu Jicong 已提交
1374
  int8_t           async = pParam->async;
L
Liu Jicong 已提交
1375
  pParam->code = code;
1376
  if (code != 0) {
L
Liu Jicong 已提交
1377
    tscError("consumer %ld get topic endpoint error, not ready, wait:%d", tmq->consumerId, pParam->async);
L
Liu Jicong 已提交
1378
    goto END;
1379
  }
L
Liu Jicong 已提交
1380

L
Liu Jicong 已提交
1381
  // tmq's epoch is monotonically increase,
L
Liu Jicong 已提交
1382
  // so it's safe to discard any old epoch msg.
L
Liu Jicong 已提交
1383
  // Epoch will only increase when received newer epoch ep msg
L
Liu Jicong 已提交
1384 1385
  SMqRspHead* head = pMsg->pData;
  int32_t     epoch = atomic_load_32(&tmq->epoch);
L
temp  
Liu Jicong 已提交
1386
  tscDebug("consumer %ld recv ep, msg epoch %d, current epoch %d", tmq->consumerId, head->epoch, epoch);
L
Liu Jicong 已提交
1387 1388
  if (head->epoch <= epoch) {
    goto END;
1389
  }
L
Liu Jicong 已提交
1390

L
Liu Jicong 已提交
1391
  if (!async) {
L
Liu Jicong 已提交
1392 1393
    SMqAskEpRsp rsp;
    tDecodeSMqAskEpRsp(POINTER_SHIFT(pMsg->pData, sizeof(SMqRspHead)), &rsp);
X
Xiaoyu Wang 已提交
1394 1395
    /*printf("rsp epoch %ld sz %ld\n", rsp.epoch, rsp.topics->size);*/
    /*printf("tmq epoch %ld sz %ld\n", tmq->epoch, tmq->clientTopics->size);*/
L
Liu Jicong 已提交
1396
    tmqUpdateEp2(tmq, head->epoch, &rsp);
L
Liu Jicong 已提交
1397
    tDeleteSMqAskEpRsp(&rsp);
X
Xiaoyu Wang 已提交
1398
  } else {
1399
    SMqAskEpRspWrapper* pWrapper = taosAllocateQitem(sizeof(SMqAskEpRspWrapper), DEF_QITEM);
L
Liu Jicong 已提交
1400
    if (pWrapper == NULL) {
X
Xiaoyu Wang 已提交
1401
      terrno = TSDB_CODE_OUT_OF_MEMORY;
L
Liu Jicong 已提交
1402 1403
      code = -1;
      goto END;
X
Xiaoyu Wang 已提交
1404
    }
L
Liu Jicong 已提交
1405 1406 1407
    pWrapper->tmqRspType = TMQ_MSG_TYPE__EP_RSP;
    pWrapper->epoch = head->epoch;
    memcpy(&pWrapper->msg, pMsg->pData, sizeof(SMqRspHead));
L
Liu Jicong 已提交
1408
    tDecodeSMqAskEpRsp(POINTER_SHIFT(pMsg->pData, sizeof(SMqRspHead)), &pWrapper->msg);
L
Liu Jicong 已提交
1409

L
Liu Jicong 已提交
1410
    taosWriteQitem(tmq->mqueue, pWrapper);
1411
    tsem_post(&tmq->rspSem);
1412
  }
L
Liu Jicong 已提交
1413 1414

END:
L
Liu Jicong 已提交
1415
  /*atomic_store_8(&tmq->epStatus, 0);*/
L
Liu Jicong 已提交
1416
  if (!async) {
L
Liu Jicong 已提交
1417
    tsem_post(&pParam->rspSem);
L
Liu Jicong 已提交
1418 1419
  } else {
    taosMemoryFree(pParam);
L
Liu Jicong 已提交
1420 1421
  }
  return code;
1422 1423
}

L
Liu Jicong 已提交
1424
int32_t tmqAskEp(tmq_t* tmq, bool async) {
L
Liu Jicong 已提交
1425
  int32_t code = 0;
L
Liu Jicong 已提交
1426
#if 0
L
Liu Jicong 已提交
1427
  int8_t  epStatus = atomic_val_compare_exchange_8(&tmq->epStatus, 0, 1);
L
fix txn  
Liu Jicong 已提交
1428
  if (epStatus == 1) {
L
temp  
Liu Jicong 已提交
1429
    int32_t epSkipCnt = atomic_add_fetch_32(&tmq->epSkipCnt, 1);
L
Liu Jicong 已提交
1430
    tscTrace("consumer %ld skip ask ep cnt %d", tmq->consumerId, epSkipCnt);
L
Liu Jicong 已提交
1431
    if (epSkipCnt < 5000) return 0;
L
fix txn  
Liu Jicong 已提交
1432
  }
L
temp  
Liu Jicong 已提交
1433
  atomic_store_32(&tmq->epSkipCnt, 0);
L
Liu Jicong 已提交
1434
#endif
L
Liu Jicong 已提交
1435
  int32_t      tlen = sizeof(SMqAskEpReq);
L
Liu Jicong 已提交
1436
  SMqAskEpReq* req = taosMemoryCalloc(1, tlen);
L
Liu Jicong 已提交
1437
  if (req == NULL) {
L
Liu Jicong 已提交
1438
    tscError("failed to malloc get subscribe ep buf");
L
Liu Jicong 已提交
1439
    /*atomic_store_8(&tmq->epStatus, 0);*/
L
Liu Jicong 已提交
1440
    return -1;
L
Liu Jicong 已提交
1441
  }
L
Liu Jicong 已提交
1442 1443 1444
  req->consumerId = htobe64(tmq->consumerId);
  req->epoch = htonl(tmq->epoch);
  strcpy(req->cgroup, tmq->groupId);
1445

L
Liu Jicong 已提交
1446
  SMqAskEpCbParam* pParam = taosMemoryCalloc(1, sizeof(SMqAskEpCbParam));
L
Liu Jicong 已提交
1447 1448
  if (pParam == NULL) {
    tscError("failed to malloc subscribe param");
wafwerar's avatar
wafwerar 已提交
1449
    taosMemoryFree(req);
L
Liu Jicong 已提交
1450
    /*atomic_store_8(&tmq->epStatus, 0);*/
L
Liu Jicong 已提交
1451
    return -1;
L
Liu Jicong 已提交
1452 1453
  }
  pParam->tmq = tmq;
L
Liu Jicong 已提交
1454
  pParam->async = async;
X
Xiaoyu Wang 已提交
1455
  tsem_init(&pParam->rspSem, 0, 0);
1456

1457
  SMsgSendInfo* sendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
L
Liu Jicong 已提交
1458 1459
  if (sendInfo == NULL) {
    tsem_destroy(&pParam->rspSem);
wafwerar's avatar
wafwerar 已提交
1460 1461
    taosMemoryFree(pParam);
    taosMemoryFree(req);
L
Liu Jicong 已提交
1462
    /*atomic_store_8(&tmq->epStatus, 0);*/
L
Liu Jicong 已提交
1463 1464 1465 1466 1467 1468 1469 1470 1471 1472
    return -1;
  }

  sendInfo->msgInfo = (SDataBuf){
      .pData = req,
      .len = tlen,
      .handle = NULL,
  };

  sendInfo->requestId = generateRequestId();
L
Liu Jicong 已提交
1473 1474 1475
  sendInfo->requestObjRefId = 0;
  sendInfo->param = pParam;
  sendInfo->fp = tmqAskEpCb;
L
Liu Jicong 已提交
1476
  sendInfo->msgType = TDMT_MND_MQ_ASK_EP;
1477

L
Liu Jicong 已提交
1478 1479
  SEpSet epSet = getEpSet_s(&tmq->pTscObj->pAppInfo->mgmtEp);

L
add log  
Liu Jicong 已提交
1480 1481
  tscDebug("consumer %ld ask ep", tmq->consumerId);

L
Liu Jicong 已提交
1482 1483
  int64_t transporterId = 0;
  asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &epSet, &transporterId, sendInfo);
1484

L
Liu Jicong 已提交
1485
  if (!async) {
L
Liu Jicong 已提交
1486 1487 1488 1489 1490
    tsem_wait(&pParam->rspSem);
    code = pParam->code;
    taosMemoryFree(pParam);
  }
  return code;
1491 1492
}

L
Liu Jicong 已提交
1493 1494
#if 0
int32_t tmq_seek(tmq_t* tmq, const tmq_topic_vgroup_t* offset) {
L
Liu Jicong 已提交
1495 1496 1497 1498 1499 1500 1501 1502 1503 1504 1505 1506 1507
  const SMqOffset* pOffset = &offset->offset;
  if (strcmp(pOffset->cgroup, tmq->groupId) != 0) {
    return TMQ_RESP_ERR__FAIL;
  }
  int32_t sz = taosArrayGetSize(tmq->clientTopics);
  for (int32_t i = 0; i < sz; i++) {
    SMqClientTopic* clientTopic = taosArrayGet(tmq->clientTopics, i);
    if (strcmp(clientTopic->topicName, pOffset->topicName) == 0) {
      int32_t vgSz = taosArrayGetSize(clientTopic->vgs);
      for (int32_t j = 0; j < vgSz; j++) {
        SMqClientVg* pVg = taosArrayGet(clientTopic->vgs, j);
        if (pVg->vgId == pOffset->vgId) {
          pVg->currentOffset = pOffset->offset;
L
Liu Jicong 已提交
1508
          tmqClearUnhandleMsg(tmq);
L
Liu Jicong 已提交
1509 1510 1511 1512 1513 1514 1515
          return TMQ_RESP_ERR__SUCCESS;
        }
      }
    }
  }
  return TMQ_RESP_ERR__FAIL;
}
L
Liu Jicong 已提交
1516
#endif
L
Liu Jicong 已提交
1517

1518
SMqPollReq* tmqBuildConsumeReqImpl(tmq_t* tmq, int64_t timeout, SMqClientTopic* pTopic, SMqClientVg* pVg) {
L
Liu Jicong 已提交
1519 1520 1521 1522
  int64_t reqOffset;
  if (pVg->currentOffset >= 0) {
    reqOffset = pVg->currentOffset;
  } else {
L
Liu Jicong 已提交
1523 1524 1525 1526
    /*if (tmq->resetOffsetCfg == TMQ_CONF__RESET_OFFSET__NONE) {*/
    /*tscError("unable to poll since no committed offset but reset offset is set to none");*/
    /*return NULL;*/
    /*}*/
L
Liu Jicong 已提交
1527 1528 1529
    reqOffset = tmq->resetOffsetCfg;
  }

L
Liu Jicong 已提交
1530
  SMqPollReq* pReq = taosMemoryCalloc(1, sizeof(SMqPollReq));
1531 1532 1533
  if (pReq == NULL) {
    return NULL;
  }
L
Liu Jicong 已提交
1534

L
Liu Jicong 已提交
1535 1536 1537
  /*strcpy(pReq->topic, pTopic->topicName);*/
  /*strcpy(pReq->cgroup, tmq->groupId);*/

L
Liu Jicong 已提交
1538 1539 1540 1541
  int32_t groupLen = strlen(tmq->groupId);
  memcpy(pReq->subKey, tmq->groupId, groupLen);
  pReq->subKey[groupLen] = TMQ_SEPARATOR;
  strcpy(pReq->subKey + groupLen + 1, pTopic->topicName);
1542

1543
  pReq->withTbName = tmq->withTbName;
1544
  pReq->timeout = timeout;
L
Liu Jicong 已提交
1545
  pReq->consumerId = tmq->consumerId;
X
Xiaoyu Wang 已提交
1546
  pReq->epoch = tmq->epoch;
L
Liu Jicong 已提交
1547
  pReq->currentOffset = reqOffset;
L
Liu Jicong 已提交
1548
  pReq->reqId = generateRequestId();
1549

L
Liu Jicong 已提交
1550 1551
  pReq->useSnapshot = tmq->useSnapshot;

1552
  pReq->head.vgId = htonl(pVg->vgId);
L
Liu Jicong 已提交
1553
  pReq->head.contLen = htonl(sizeof(SMqPollReq));
1554 1555 1556
  return pReq;
}

L
Liu Jicong 已提交
1557 1558
SMqMetaRspObj* tmqBuildMetaRspFromWrapper(SMqPollRspWrapper* pWrapper) {
  SMqMetaRspObj* pRspObj = taosMemoryCalloc(1, sizeof(SMqMetaRspObj));
L
Liu Jicong 已提交
1559
  pRspObj->resType = RES_TYPE__TMQ_META;
L
Liu Jicong 已提交
1560 1561 1562 1563 1564 1565 1566 1567
  tstrncpy(pRspObj->topic, pWrapper->topicHandle->topicName, TSDB_TOPIC_FNAME_LEN);
  tstrncpy(pRspObj->db, pWrapper->topicHandle->db, TSDB_DB_FNAME_LEN);
  pRspObj->vgId = pWrapper->vgHandle->vgId;

  memcpy(&pRspObj->metaRsp, &pWrapper->metaRsp, sizeof(SMqMetaRsp));
  return pRspObj;
}

L
Liu Jicong 已提交
1568 1569 1570
SMqRspObj* tmqBuildRspFromWrapper(SMqPollRspWrapper* pWrapper) {
  SMqRspObj* pRspObj = taosMemoryCalloc(1, sizeof(SMqRspObj));
  pRspObj->resType = RES_TYPE__TMQ;
L
Liu Jicong 已提交
1571 1572
  tstrncpy(pRspObj->topic, pWrapper->topicHandle->topicName, TSDB_TOPIC_FNAME_LEN);
  tstrncpy(pRspObj->db, pWrapper->topicHandle->db, TSDB_DB_FNAME_LEN);
L
Liu Jicong 已提交
1573
  pRspObj->vgId = pWrapper->vgHandle->vgId;
L
Liu Jicong 已提交
1574
  pRspObj->resIter = -1;
L
Liu Jicong 已提交
1575
  memcpy(&pRspObj->rsp, &pWrapper->dataRsp, sizeof(SMqDataBlkRsp));
L
Liu Jicong 已提交
1576

L
Liu Jicong 已提交
1577 1578
  pRspObj->resInfo.totalRows = 0;
  pRspObj->resInfo.precision = TSDB_TIME_PRECISION_MILLI;
L
Liu Jicong 已提交
1579
  if (!pWrapper->dataRsp.withSchema) {
L
Liu Jicong 已提交
1580 1581
    setResSchemaInfo(&pRspObj->resInfo, pWrapper->topicHandle->schema.pSchema, pWrapper->topicHandle->schema.nCols);
  }
L
Liu Jicong 已提交
1582

L
Liu Jicong 已提交
1583
  return pRspObj;
X
Xiaoyu Wang 已提交
1584 1585
}

1586
int32_t tmqPollImpl(tmq_t* tmq, int64_t timeout) {
L
fix  
Liu Jicong 已提交
1587
  /*printf("call poll\n");*/
X
Xiaoyu Wang 已提交
1588 1589 1590 1591 1592 1593
  for (int i = 0; i < taosArrayGetSize(tmq->clientTopics); i++) {
    SMqClientTopic* pTopic = taosArrayGet(tmq->clientTopics, i);
    for (int j = 0; j < taosArrayGetSize(pTopic->vgs); j++) {
      SMqClientVg* pVg = taosArrayGet(pTopic->vgs, j);
      int32_t      vgStatus = atomic_val_compare_exchange_32(&pVg->vgStatus, TMQ_VG_STATUS__IDLE, TMQ_VG_STATUS__WAIT);
      if (vgStatus != TMQ_VG_STATUS__IDLE) {
L
Liu Jicong 已提交
1594
        int32_t vgSkipCnt = atomic_add_fetch_32(&pVg->vgSkipCnt, 1);
L
Liu Jicong 已提交
1595
        tscTrace("consumer %ld epoch %d skip vg %d skip cnt %d", tmq->consumerId, tmq->epoch, pVg->vgId, vgSkipCnt);
X
Xiaoyu Wang 已提交
1596
        continue;
L
Liu Jicong 已提交
1597
        /*if (vgSkipCnt < 10000) continue;*/
L
temp  
Liu Jicong 已提交
1598 1599 1600 1601 1602 1603 1604
#if 0
        if (skipCnt < 30000) {
          continue;
        } else {
        tscDebug("consumer %ld skip vg %d skip too much reset", tmq->consumerId, pVg->vgId);
        }
#endif
X
Xiaoyu Wang 已提交
1605
      }
L
Liu Jicong 已提交
1606
      atomic_store_32(&pVg->vgSkipCnt, 0);
1607
      SMqPollReq* pReq = tmqBuildConsumeReqImpl(tmq, timeout, pTopic, pVg);
X
Xiaoyu Wang 已提交
1608 1609
      if (pReq == NULL) {
        atomic_store_32(&pVg->vgStatus, TMQ_VG_STATUS__IDLE);
1610
        tsem_post(&tmq->rspSem);
X
Xiaoyu Wang 已提交
1611 1612
        return -1;
      }
wafwerar's avatar
wafwerar 已提交
1613
      SMqPollCbParam* pParam = taosMemoryMalloc(sizeof(SMqPollCbParam));
L
Liu Jicong 已提交
1614
      if (pParam == NULL) {
wafwerar's avatar
wafwerar 已提交
1615
        taosMemoryFree(pReq);
X
Xiaoyu Wang 已提交
1616
        atomic_store_32(&pVg->vgStatus, TMQ_VG_STATUS__IDLE);
1617
        tsem_post(&tmq->rspSem);
X
Xiaoyu Wang 已提交
1618 1619
        return -1;
      }
L
Liu Jicong 已提交
1620 1621
      pParam->tmq = tmq;
      pParam->pVg = pVg;
L
Liu Jicong 已提交
1622
      pParam->pTopic = pTopic;
L
Liu Jicong 已提交
1623
      pParam->vgId = pVg->vgId;
L
Liu Jicong 已提交
1624 1625
      pParam->epoch = tmq->epoch;

1626
      SMsgSendInfo* sendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
L
Liu Jicong 已提交
1627
      if (sendInfo == NULL) {
wafwerar's avatar
wafwerar 已提交
1628 1629
        taosMemoryFree(pReq);
        taosMemoryFree(pParam);
L
Liu Jicong 已提交
1630
        atomic_store_32(&pVg->vgStatus, TMQ_VG_STATUS__IDLE);
1631
        tsem_post(&tmq->rspSem);
L
Liu Jicong 已提交
1632 1633 1634 1635
        return -1;
      }

      sendInfo->msgInfo = (SDataBuf){
X
Xiaoyu Wang 已提交
1636
          .pData = pReq,
L
Liu Jicong 已提交
1637
          .len = sizeof(SMqPollReq),
X
Xiaoyu Wang 已提交
1638 1639
          .handle = NULL,
      };
L
Liu Jicong 已提交
1640
      sendInfo->requestId = pReq->reqId;
X
Xiaoyu Wang 已提交
1641
      sendInfo->requestObjRefId = 0;
L
Liu Jicong 已提交
1642
      sendInfo->param = pParam;
X
Xiaoyu Wang 已提交
1643
      sendInfo->fp = tmqPollCb;
L
Liu Jicong 已提交
1644
      sendInfo->msgType = TDMT_VND_CONSUME;
X
Xiaoyu Wang 已提交
1645 1646

      int64_t transporterId = 0;
L
fix  
Liu Jicong 已提交
1647
      /*printf("send poll\n");*/
L
Liu Jicong 已提交
1648 1649
      tscDebug("consumer %ld send poll to %s : vg %d, epoch %d, req offset %ld, reqId %lu", tmq->consumerId,
               pTopic->topicName, pVg->vgId, tmq->epoch, pVg->currentOffset, pReq->reqId);
L
fix  
Liu Jicong 已提交
1650
      /*printf("send vg %d %ld\n", pVg->vgId, pVg->currentOffset);*/
X
Xiaoyu Wang 已提交
1651 1652 1653 1654 1655 1656 1657 1658
      asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &pVg->epSet, &transporterId, sendInfo);
      pVg->pollCnt++;
      tmq->pollCnt++;
    }
  }
  return 0;
}

L
Liu Jicong 已提交
1659 1660
int32_t tmqHandleNoPollRsp(tmq_t* tmq, SMqRspWrapper* rspWrapper, bool* pReset) {
  if (rspWrapper->tmqRspType == TMQ_MSG_TYPE__EP_RSP) {
L
fix  
Liu Jicong 已提交
1661
    /*printf("ep %d %d\n", rspMsg->head.epoch, tmq->epoch);*/
L
Liu Jicong 已提交
1662 1663
    if (rspWrapper->epoch > atomic_load_32(&tmq->epoch)) {
      SMqAskEpRspWrapper* pEpRspWrapper = (SMqAskEpRspWrapper*)rspWrapper;
L
Liu Jicong 已提交
1664
      SMqAskEpRsp*        rspMsg = &pEpRspWrapper->msg;
L
Liu Jicong 已提交
1665
      tmqUpdateEp2(tmq, rspWrapper->epoch, rspMsg);
L
temp  
Liu Jicong 已提交
1666
      /*tmqClearUnhandleMsg(tmq);*/
X
Xiaoyu Wang 已提交
1667 1668 1669 1670 1671 1672 1673 1674 1675 1676
      *pReset = true;
    } else {
      *pReset = false;
    }
  } else {
    return -1;
  }
  return 0;
}

L
Liu Jicong 已提交
1677
void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) {
X
Xiaoyu Wang 已提交
1678
  while (1) {
L
Liu Jicong 已提交
1679 1680 1681
    SMqRspWrapper* rspWrapper = NULL;
    taosGetQitem(tmq->qall, (void**)&rspWrapper);
    if (rspWrapper == NULL) {
L
Liu Jicong 已提交
1682
      taosReadAllQitems(tmq->mqueue, tmq->qall);
L
Liu Jicong 已提交
1683 1684
      taosGetQitem(tmq->qall, (void**)&rspWrapper);
      if (rspWrapper == NULL) return NULL;
X
Xiaoyu Wang 已提交
1685 1686
    }

L
Liu Jicong 已提交
1687 1688 1689 1690 1691
    if (rspWrapper->tmqRspType == TMQ_MSG_TYPE__END_RSP) {
      taosFreeQitem(rspWrapper);
      terrno = TSDB_CODE_TQ_NO_COMMITTED_OFFSET;
      return NULL;
    } else if (rspWrapper->tmqRspType == TMQ_MSG_TYPE__POLL_RSP) {
L
Liu Jicong 已提交
1692
      SMqPollRspWrapper* pollRspWrapper = (SMqPollRspWrapper*)rspWrapper;
L
Liu Jicong 已提交
1693
      /*atomic_sub_fetch_32(&tmq->readyRequest, 1);*/
1694
      int32_t consumerEpoch = atomic_load_32(&tmq->epoch);
L
Liu Jicong 已提交
1695
      if (pollRspWrapper->dataRsp.head.epoch == consumerEpoch) {
L
Liu Jicong 已提交
1696
        SMqClientVg* pVg = pollRspWrapper->vgHandle;
L
Liu Jicong 已提交
1697
        /*printf("vg %d offset %ld up to %ld\n", pVg->vgId, pVg->currentOffset, rspMsg->msg.rspOffset);*/
L
Liu Jicong 已提交
1698
        pVg->currentOffset = pollRspWrapper->dataRsp.rspOffset;
X
Xiaoyu Wang 已提交
1699
        atomic_store_32(&pVg->vgStatus, TMQ_VG_STATUS__IDLE);
L
Liu Jicong 已提交
1700
        if (pollRspWrapper->dataRsp.blockNum == 0) {
L
Liu Jicong 已提交
1701 1702
          taosFreeQitem(pollRspWrapper);
          rspWrapper = NULL;
L
temp  
Liu Jicong 已提交
1703 1704
          continue;
        }
L
Liu Jicong 已提交
1705
        // build rsp
L
Liu Jicong 已提交
1706
        SMqRspObj* pRsp = tmqBuildRspFromWrapper(pollRspWrapper);
L
Liu Jicong 已提交
1707
        taosFreeQitem(pollRspWrapper);
L
Liu Jicong 已提交
1708
        return pRsp;
X
Xiaoyu Wang 已提交
1709
      } else {
L
Liu Jicong 已提交
1710 1711 1712 1713 1714 1715 1716
        tscDebug("msg discard since epoch mismatch: msg epoch %d, consumer epoch %d\n",
                 pollRspWrapper->dataRsp.head.epoch, consumerEpoch);
        taosFreeQitem(pollRspWrapper);
      }
    } else if (rspWrapper->tmqRspType == TMQ_MSG_TYPE__POLL_META_RSP) {
      SMqPollRspWrapper* pollRspWrapper = (SMqPollRspWrapper*)rspWrapper;
      int32_t            consumerEpoch = atomic_load_32(&tmq->epoch);
L
Liu Jicong 已提交
1717
      if (pollRspWrapper->metaRsp.head.epoch == consumerEpoch) {
L
Liu Jicong 已提交
1718 1719
        SMqClientVg* pVg = pollRspWrapper->vgHandle;
        /*printf("vg %d offset %ld up to %ld\n", pVg->vgId, pVg->currentOffset, rspMsg->msg.rspOffset);*/
L
Liu Jicong 已提交
1720
        pVg->currentOffset = pollRspWrapper->metaRsp.rspOffset;
L
Liu Jicong 已提交
1721 1722
        atomic_store_32(&pVg->vgStatus, TMQ_VG_STATUS__IDLE);
        // build rsp
L
Liu Jicong 已提交
1723
        SMqMetaRspObj* pRsp = tmqBuildMetaRspFromWrapper(pollRspWrapper);
L
Liu Jicong 已提交
1724 1725 1726 1727
        taosFreeQitem(pollRspWrapper);
        return pRsp;
      } else {
        tscDebug("msg discard since epoch mismatch: msg epoch %d, consumer epoch %d\n",
L
Liu Jicong 已提交
1728
                 pollRspWrapper->metaRsp.head.epoch, consumerEpoch);
L
Liu Jicong 已提交
1729
        taosFreeQitem(pollRspWrapper);
X
Xiaoyu Wang 已提交
1730 1731
      }
    } else {
L
fix  
Liu Jicong 已提交
1732
      /*printf("handle ep rsp %d\n", rspMsg->head.mqMsgType);*/
X
Xiaoyu Wang 已提交
1733
      bool reset = false;
L
Liu Jicong 已提交
1734 1735
      tmqHandleNoPollRsp(tmq, rspWrapper, &reset);
      taosFreeQitem(rspWrapper);
X
Xiaoyu Wang 已提交
1736
      if (pollIfReset && reset) {
L
add log  
Liu Jicong 已提交
1737
        tscDebug("consumer %ld reset and repoll", tmq->consumerId);
1738
        tmqPollImpl(tmq, timeout);
X
Xiaoyu Wang 已提交
1739 1740 1741 1742 1743
      }
    }
  }
}

1744
TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t timeout) {
L
Liu Jicong 已提交
1745 1746
  void*   rspObj;
  int64_t startTime = taosGetTimestampMs();
L
Liu Jicong 已提交
1747

1748 1749 1750
#if 0
  tmqHandleAllDelayedTask(tmq);
  tmqPollImpl(tmq, timeout);
1751
  rspObj = tmqHandleAllRsp(tmq, timeout, false);
L
Liu Jicong 已提交
1752 1753
  if (rspObj) {
    return (TAOS_RES*)rspObj;
L
fix  
Liu Jicong 已提交
1754
  }
1755
#endif
X
Xiaoyu Wang 已提交
1756

L
Liu Jicong 已提交
1757
  // in no topic status also need process delayed task
L
Liu Jicong 已提交
1758
  if (atomic_load_8(&tmq->status) == TMQ_CONSUMER_STATUS__INIT) {
1759 1760 1761
    return NULL;
  }

X
Xiaoyu Wang 已提交
1762
  while (1) {
L
Liu Jicong 已提交
1763
    tmqHandleAllDelayedTask(tmq);
1764
    if (tmqPollImpl(tmq, timeout) < 0) return NULL;
L
Liu Jicong 已提交
1765

1766
    rspObj = tmqHandleAllRsp(tmq, timeout, false);
L
Liu Jicong 已提交
1767 1768
    if (rspObj) {
      return (TAOS_RES*)rspObj;
L
Liu Jicong 已提交
1769 1770
    } else if (terrno == TSDB_CODE_TQ_NO_COMMITTED_OFFSET) {
      return NULL;
X
Xiaoyu Wang 已提交
1771
    }
1772
    if (timeout != -1) {
X
Xiaoyu Wang 已提交
1773
      int64_t endTime = taosGetTimestampMs();
1774
      int64_t leftTime = endTime - startTime;
1775
      if (leftTime > timeout) {
L
Liu Jicong 已提交
1776
        tscDebug("consumer %ld (epoch %d) timeout, no rsp", tmq->consumerId, tmq->epoch);
X
Xiaoyu Wang 已提交
1777 1778
        return NULL;
      }
1779
      tsem_timewait(&tmq->rspSem, leftTime * 1000);
L
Liu Jicong 已提交
1780 1781 1782
    } else {
      // use tsem_timewait instead of tsem_wait to avoid unexpected stuck
      tsem_timewait(&tmq->rspSem, 500 * 1000);
X
Xiaoyu Wang 已提交
1783 1784 1785 1786
    }
  }
}

L
Liu Jicong 已提交
1787
int32_t tmq_consumer_close(tmq_t* tmq) {
1788
  if (tmq->status == TMQ_CONSUMER_STATUS__READY) {
L
Liu Jicong 已提交
1789 1790
    int32_t rsp = tmq_commit_sync(tmq, NULL);
    if (rsp != 0) {
L
Liu Jicong 已提交
1791
      return rsp;
1792 1793 1794 1795
    }

    tmq_list_t* lst = tmq_list_new();
    rsp = tmq_subscribe(tmq, lst);
1796
    tmq_list_destroy(lst);
1797

L
Liu Jicong 已提交
1798
    if (rsp != 0) {
L
Liu Jicong 已提交
1799
      return rsp;
1800
    }
L
Liu Jicong 已提交
1801
  }
1802
  // TODO: free resources
L
Liu Jicong 已提交
1803
  return 0;
1804
}
L
Liu Jicong 已提交
1805

L
Liu Jicong 已提交
1806 1807
const char* tmq_err2str(int32_t err) {
  if (err == 0) {
L
Liu Jicong 已提交
1808
    return "success";
L
Liu Jicong 已提交
1809
  } else if (err == -1) {
L
Liu Jicong 已提交
1810 1811 1812
    return "fail";
  } else {
    return tstrerror(err);
L
Liu Jicong 已提交
1813 1814
  }
}
L
Liu Jicong 已提交
1815

L
Liu Jicong 已提交
1816 1817 1818 1819 1820 1821 1822 1823 1824 1825
tmq_res_t tmq_get_res_type(TAOS_RES* res) {
  if (TD_RES_TMQ(res)) {
    return TMQ_RES_DATA;
  } else if (TD_RES_TMQ_META(res)) {
    return TMQ_RES_TABLE_META;
  } else {
    return TMQ_RES_INVALID;
  }
}

L
Liu Jicong 已提交
1826
const char* tmq_get_topic_name(TAOS_RES* res) {
L
Liu Jicong 已提交
1827 1828
  if (TD_RES_TMQ(res)) {
    SMqRspObj* pRspObj = (SMqRspObj*)res;
L
Liu Jicong 已提交
1829
    return strchr(pRspObj->topic, '.') + 1;
L
Liu Jicong 已提交
1830 1831 1832
  } else if (TD_RES_TMQ_META(res)) {
    SMqMetaRspObj* pMetaRspObj = (SMqMetaRspObj*)res;
    return strchr(pMetaRspObj->topic, '.') + 1;
L
Liu Jicong 已提交
1833 1834 1835 1836 1837
  } else {
    return NULL;
  }
}

L
Liu Jicong 已提交
1838 1839 1840 1841
const char* tmq_get_db_name(TAOS_RES* res) {
  if (TD_RES_TMQ(res)) {
    SMqRspObj* pRspObj = (SMqRspObj*)res;
    return strchr(pRspObj->db, '.') + 1;
L
Liu Jicong 已提交
1842 1843 1844
  } else if (TD_RES_TMQ_META(res)) {
    SMqMetaRspObj* pMetaRspObj = (SMqMetaRspObj*)res;
    return strchr(pMetaRspObj->db, '.') + 1;
L
Liu Jicong 已提交
1845 1846 1847 1848 1849
  } else {
    return NULL;
  }
}

L
Liu Jicong 已提交
1850 1851 1852 1853
int32_t tmq_get_vgroup_id(TAOS_RES* res) {
  if (TD_RES_TMQ(res)) {
    SMqRspObj* pRspObj = (SMqRspObj*)res;
    return pRspObj->vgId;
L
Liu Jicong 已提交
1854 1855 1856
  } else if (TD_RES_TMQ_META(res)) {
    SMqMetaRspObj* pMetaRspObj = (SMqMetaRspObj*)res;
    return pMetaRspObj->vgId;
L
Liu Jicong 已提交
1857 1858 1859 1860
  } else {
    return -1;
  }
}
L
Liu Jicong 已提交
1861 1862 1863 1864 1865 1866 1867 1868

const char* tmq_get_table_name(TAOS_RES* res) {
  if (TD_RES_TMQ(res)) {
    SMqRspObj* pRspObj = (SMqRspObj*)res;
    if (!pRspObj->rsp.withTbName || pRspObj->rsp.blockTbName == NULL || pRspObj->resIter < 0 ||
        pRspObj->resIter >= pRspObj->rsp.blockNum) {
      return NULL;
    }
1869
    return (const char*)taosArrayGetP(pRspObj->rsp.blockTbName, pRspObj->resIter);
L
Liu Jicong 已提交
1870 1871 1872
  }
  return NULL;
}
1873

1874
int32_t tmq_get_raw_meta(TAOS_RES* res, void** raw_meta, int32_t* raw_meta_len) {
L
Liu Jicong 已提交
1875 1876 1877 1878 1879 1880 1881 1882 1883
  if (TD_RES_TMQ_META(res)) {
    SMqMetaRspObj* pMetaRspObj = (SMqMetaRspObj*)res;
    *raw_meta = pMetaRspObj->metaRsp.metaRsp;
    *raw_meta_len = pMetaRspObj->metaRsp.metaRspLen;
    return 0;
  }
  return -1;
}

L
Liu Jicong 已提交
1884 1885
void tmq_commit_async(tmq_t* tmq, const TAOS_RES* msg, tmq_commit_cb* cb, void* param) {
  tmqCommitInner2(tmq, msg, 0, 1, cb, param);
L
Liu Jicong 已提交
1886 1887
}

L
Liu Jicong 已提交
1888
int32_t tmq_commit_sync(tmq_t* tmq, const TAOS_RES* msg) { return tmqCommitInner2(tmq, msg, 0, 0, NULL, NULL); }