diff --git a/examples/c/tmq.c b/examples/c/tmq.c index 94e0b86821b4dc601709b4b513027b5ccf882861..38c3eba647285ce29e3778fc0b85ae84944c2f67 100644 --- a/examples/c/tmq.c +++ b/examples/c/tmq.c @@ -15,10 +15,10 @@ #include #include +#include #include #include #include "taos.h" -#include static int running = 1; static void msg_process(TAOS_RES* msg) { @@ -28,8 +28,8 @@ static void msg_process(TAOS_RES* msg) { printf("db: %s\n", tmq_get_db_name(msg)); printf("vg: %d\n", tmq_get_vgroup_id(msg)); if (tmq_get_res_type(msg) == TMQ_RES_TABLE_META) { - tmq_raw_data *raw = tmq_get_raw_meta(msg); - if(raw){ + tmq_raw_data* raw = tmq_get_raw_meta(msg); + if (raw) { TAOS* pConn = taos_connect("192.168.1.86", "root", "taosdata", NULL, 0); if (pConn == NULL) { return; @@ -55,7 +55,7 @@ static void msg_process(TAOS_RES* msg) { } tmq_free_raw_meta(raw); char* result = tmq_get_json_meta(msg); - if(result){ + if (result) { printf("meta result: %s\n", result); } tmq_free_json_meta(result); @@ -96,7 +96,9 @@ int32_t init_env() { } taos_free_result(pRes); - pRes = taos_query(pConn, "create stable if not exists st1 (ts timestamp, c1 int, c2 float, c3 binary(16)) tags(t1 int, t3 nchar(8), t4 bool)"); + pRes = taos_query(pConn, + "create stable if not exists st1 (ts timestamp, c1 int, c2 float, c3 binary(16)) tags(t1 int, t3 " + "nchar(8), t4 bool)"); if (taos_errno(pRes) != 0) { printf("failed to create super table st1, reason:%s\n", taos_errstr(pRes)); return -1; @@ -152,6 +154,7 @@ int32_t init_env() { } taos_free_result(pRes); +#if 0 pRes = taos_query(pConn, "alter table st1 add column c4 bigint"); if (taos_errno(pRes) != 0) { printf("failed to alter super table st1, reason:%s\n", taos_errstr(pRes)); @@ -264,7 +267,9 @@ int32_t init_env() { } taos_free_result(pRes); - pRes = taos_query(pConn, "create stable if not exists st1 (ts timestamp, c1 int, c2 float, c3 binary(16)) tags(t1 int, t3 nchar(8), t4 bool)"); + pRes = taos_query(pConn, + "create stable if not exists st1 (ts timestamp, c1 int, c2 float, c3 binary(16)) tags(t1 int, t3 " + "nchar(8), t4 bool)"); if (taos_errno(pRes) != 0) { printf("failed to create super table st1, reason:%s\n", taos_errstr(pRes)); return -1; @@ -277,6 +282,7 @@ int32_t init_env() { return -1; } taos_free_result(pRes); +#endif return 0; } @@ -296,8 +302,15 @@ int32_t create_topic() { } taos_free_result(pRes); - pRes = taos_query(pConn, "create topic topic_ctb_column with meta as database abc1"); - /*pRes = taos_query(pConn, "create topic topic_ctb_column as select ts, c1, c2, c3 from st1");*/ + /*pRes = taos_query(pConn, "create topic topic_ctb_column with meta as database abc1");*/ + pRes = taos_query(pConn, "create topic topic_ctb_column as select ts, c1, c2, c3 from st1"); + if (taos_errno(pRes) != 0) { + printf("failed to create topic topic_ctb_column, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "create topic topic2 as select ts, c1, c2, c3 from st1"); if (taos_errno(pRes) != 0) { printf("failed to create topic topic_ctb_column, reason:%s\n", taos_errstr(pRes)); return -1; @@ -353,6 +366,7 @@ tmq_t* build_consumer() { tmq_conf_t* conf = tmq_conf_new(); tmq_conf_set(conf, "group.id", "tg2"); + tmq_conf_set(conf, "client.id", "my app 1"); tmq_conf_set(conf, "td.connect.user", "root"); tmq_conf_set(conf, "td.connect.pass", "taosdata"); tmq_conf_set(conf, "msg.with.table.name", "true"); diff --git a/include/common/tmsg.h b/include/common/tmsg.h index 61ae7541a025d9899f5ac6b1da05473fb9e2d555..9bdfacbf3628e9dab716f28ab5538bbd9ebd7f5b 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -2923,7 +2923,6 @@ typedef struct { SMqRspHead head; STqOffsetVal reqOffset; STqOffsetVal rspOffset; - int32_t skipLogNum; int32_t blockNum; int8_t withTbName; int8_t withSchema; diff --git a/include/libs/stream/tstream.h b/include/libs/stream/tstream.h index 5c4d8ce2500e2cba34806045900d5a4287b9bec8..8c69c0f2de498fcf463e0e2d8624db29cc610772 100644 --- a/include/libs/stream/tstream.h +++ b/include/libs/stream/tstream.h @@ -83,6 +83,7 @@ typedef struct { int32_t srcVgId; int32_t childId; int64_t sourceVer; + int64_t reqId; SArray* blocks; // SArray } SStreamDataBlock; @@ -324,6 +325,8 @@ static FORCE_INLINE int32_t streamTaskInput(SStreamTask* pTask, SStreamQueueItem if (pItem->type == STREAM_INPUT__DATA_SUBMIT) { SStreamDataSubmit* pSubmitClone = streamSubmitRefClone((SStreamDataSubmit*)pItem); if (pSubmitClone == NULL) { + qDebug("task %d %p submit enqueue failed since out of memory", pTask->taskId, pTask); + terrno = TSDB_CODE_OUT_OF_MEMORY; atomic_store_8(&pTask->inputStatus, TASK_INPUT_STATUS__FAILED); return -1; } @@ -412,6 +415,7 @@ typedef struct { typedef struct { int64_t streamId; + int64_t reqId; int32_t srcTaskId; int32_t srcNodeId; int32_t dstTaskId; diff --git a/source/client/src/tmq.c b/source/client/src/tmq.c index ed0ec516b280410f5ad1966c8fd4abc17dc6dda1..5b845fd455622cf5046a5949cb2845f787850988 100644 --- a/source/client/src/tmq.c +++ b/source/client/src/tmq.c @@ -1052,6 +1052,7 @@ int32_t tmq_subscribe(tmq_t* tmq, const tmq_list_t* topic_list) { int32_t code = -1; req.consumerId = tmq->consumerId; + tstrncpy(req.clientId, tmq->clientId, 256); tstrncpy(req.cgroup, tmq->groupId, TSDB_CGROUP_LEN); req.topicNames = taosArrayInit(sz, sizeof(void*)); if (req.topicNames == NULL) goto FAIL; @@ -1146,14 +1147,6 @@ void tmq_conf_set_auto_commit_cb(tmq_conf_t* conf, tmq_commit_cb* cb, void* para conf->commitCbUserParam = param; } -#if 0 -int32_t tmqGetSkipLogNum(tmq_message_t* tmq_message) { - if (tmq_message == NULL) return 0; - SMqPollRsp* pRsp = &tmq_message->msg; - return pRsp->skipLogNum; -} -#endif - int32_t tmqPollCb(void* param, SDataBuf* pMsg, int32_t code) { SMqPollCbParam* pParam = (SMqPollCbParam*)param; SMqClientVg* pVg = pParam->pVg; @@ -1296,9 +1289,6 @@ bool tmqUpdateEp2(tmq_t* tmq, int32_t epoch, SMqAskEpRsp* pRsp) { offsetNew = *pOffset; } - /*tscDebug("consumer:%" PRId64 ", (epoch %d) offset of vgId:%d updated to %" PRId64 ", vgKey is %s", - * tmq->consumerId, epoch,*/ - /*pVgEp->vgId, offset, vgKey);*/ SMqClientVg clientVg = { .pollCnt = 0, .currentOffsetNew = offsetNew, diff --git a/source/common/src/tmsg.c b/source/common/src/tmsg.c index aeb83d3425d55a06979b3135dd5a276067ca7056..8c2438eef1259d470d19cc89cec91d1b16eba1c9 100644 --- a/source/common/src/tmsg.c +++ b/source/common/src/tmsg.c @@ -5648,7 +5648,6 @@ int32_t tDecodeDeleteRes(SDecoder *pCoder, SDeleteRes *pRes) { int32_t tEncodeSMqDataRsp(SEncoder *pEncoder, const SMqDataRsp *pRsp) { if (tEncodeSTqOffsetVal(pEncoder, &pRsp->reqOffset) < 0) return -1; if (tEncodeSTqOffsetVal(pEncoder, &pRsp->rspOffset) < 0) return -1; - if (tEncodeI32(pEncoder, pRsp->skipLogNum) < 0) return -1; if (tEncodeI32(pEncoder, pRsp->blockNum) < 0) return -1; if (pRsp->blockNum != 0) { if (tEncodeI8(pEncoder, pRsp->withTbName) < 0) return -1; @@ -5674,7 +5673,6 @@ int32_t tEncodeSMqDataRsp(SEncoder *pEncoder, const SMqDataRsp *pRsp) { int32_t tDecodeSMqDataRsp(SDecoder *pDecoder, SMqDataRsp *pRsp) { if (tDecodeSTqOffsetVal(pDecoder, &pRsp->reqOffset) < 0) return -1; if (tDecodeSTqOffsetVal(pDecoder, &pRsp->rspOffset) < 0) return -1; - if (tDecodeI32(pDecoder, &pRsp->skipLogNum) < 0) return -1; if (tDecodeI32(pDecoder, &pRsp->blockNum) < 0) return -1; if (pRsp->blockNum != 0) { pRsp->blockData = taosArrayInit(pRsp->blockNum, sizeof(void *)); diff --git a/source/dnode/mnode/impl/src/mndConsumer.c b/source/dnode/mnode/impl/src/mndConsumer.c index 5509ee88bf57d6f0e2fbe90efdfa2247c1fbec68..a60db8a8c28ccc08251310c2b78c8786f1a0a9e2 100644 --- a/source/dnode/mnode/impl/src/mndConsumer.c +++ b/source/dnode/mnode/impl/src/mndConsumer.c @@ -92,7 +92,9 @@ static int32_t mndProcessConsumerLostMsg(SRpcMsg *pMsg) { SMnode *pMnode = pMsg->info.node; SMqConsumerLostMsg *pLostMsg = pMsg->pCont; SMqConsumerObj *pConsumer = mndAcquireConsumer(pMnode, pLostMsg->consumerId); - ASSERT(pConsumer); + if (pConsumer == NULL) { + return 0; + } mInfo("receive consumer lost msg, consumer id %" PRId64 ", status %s", pLostMsg->consumerId, mndConsumerStatusName(pConsumer->status)); @@ -450,6 +452,7 @@ static int32_t mndProcessSubscribeReq(SRpcMsg *pMsg) { int32_t code = -1; SArray *newSub = subscribe.topicNames; taosArraySortString(newSub, taosArrayCompareString); + taosArrayRemoveDuplicate(newSub, taosArrayCompareString, taosMemoryFree); int32_t newTopicNum = taosArrayGetSize(newSub); // check topic existance @@ -907,8 +910,8 @@ static int32_t mndRetrieveConsumer(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock * colDataAppend(pColInfo, numOfRows, (const char *)cgroup, false); // client id - char clientId[TSDB_CGROUP_LEN + VARSTR_HEADER_SIZE] = {0}; - tstrncpy(varDataVal(clientId), pConsumer->clientId, TSDB_CGROUP_LEN); + char clientId[256 + VARSTR_HEADER_SIZE] = {0}; + tstrncpy(varDataVal(clientId), pConsumer->clientId, 256); varDataSetLen(clientId, strlen(varDataVal(clientId))); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); colDataAppend(pColInfo, numOfRows, (const char *)clientId, false); diff --git a/source/dnode/mnode/impl/src/mndDef.c b/source/dnode/mnode/impl/src/mndDef.c index c26424e04910f7d1589a24e927b521a746182575..abac0573da0eea96a047b5163d3aedc09858d153 100644 --- a/source/dnode/mnode/impl/src/mndDef.c +++ b/source/dnode/mnode/impl/src/mndDef.c @@ -199,6 +199,7 @@ int32_t tEncodeSMqConsumerObj(void **buf, const SMqConsumerObj *pConsumer) { int32_t tlen = 0; int32_t sz; tlen += taosEncodeFixedI64(buf, pConsumer->consumerId); + tlen += taosEncodeString(buf, pConsumer->clientId); tlen += taosEncodeString(buf, pConsumer->cgroup); tlen += taosEncodeFixedI8(buf, pConsumer->updateType); tlen += taosEncodeFixedI32(buf, pConsumer->epoch); @@ -264,6 +265,7 @@ int32_t tEncodeSMqConsumerObj(void **buf, const SMqConsumerObj *pConsumer) { void *tDecodeSMqConsumerObj(const void *buf, SMqConsumerObj *pConsumer) { int32_t sz; buf = taosDecodeFixedI64(buf, &pConsumer->consumerId); + buf = taosDecodeStringTo(buf, pConsumer->clientId); buf = taosDecodeStringTo(buf, pConsumer->cgroup); buf = taosDecodeFixedI8(buf, &pConsumer->updateType); buf = taosDecodeFixedI32(buf, &pConsumer->epoch); diff --git a/source/dnode/vnode/src/inc/tq.h b/source/dnode/vnode/src/inc/tq.h index c62b7e95bfe4aa0d54401f5bd1b2434476db0cba..757749a9b64db316ab2cdc8204523656e1b6ab72 100644 --- a/source/dnode/vnode/src/inc/tq.h +++ b/source/dnode/vnode/src/inc/tq.h @@ -52,7 +52,6 @@ typedef struct { int64_t reqOffset; int64_t processedVer; int32_t epoch; - int32_t skipLogNum; // rpc info int64_t reqId; SRpcHandleInfo rpcInfo; diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index f36e6f874f7f27b7338acb03e0cbdec4914eed17..fb05aeecd935c0ae1c4bcb2f6247fe23cfa6bcc4 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -635,6 +635,8 @@ int32_t tqProcessStreamTrigger(STQ* pTq, SSubmitReq* pReq) { pSubmit = streamDataSubmitNew(pReq); if (pSubmit == NULL) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + qError("failed to create data submit for stream since out of memory"); failed = true; } @@ -644,12 +646,16 @@ int32_t tqProcessStreamTrigger(STQ* pTq, SSubmitReq* pReq) { SStreamTask* pTask = *(SStreamTask**)pIter; if (!pTask->isDataScan) continue; + qDebug("data submit enqueue stream task: %d", pTask->taskId); + if (!failed) { if (streamTaskInput(pTask, (SStreamQueueItem*)pSubmit) < 0) { + qError("stream task input failed, task id %d", pTask->taskId); continue; } if (streamLaunchByWrite(pTask, TD_VID(pTq->pVnode)) < 0) { + qError("stream task launch failed, task id %d", pTask->taskId); continue; } } else { diff --git a/source/dnode/vnode/src/tq/tqExec.c b/source/dnode/vnode/src/tq/tqExec.c index 0bdbe82b772e807e6e05798ecb836b3b089d1bb0..eb17f06cc98f0b7f921fc86dee2b4c2a26fe444d 100644 --- a/source/dnode/vnode/src/tq/tqExec.c +++ b/source/dnode/vnode/src/tq/tqExec.c @@ -217,7 +217,6 @@ int32_t tqLogScanExec(STQ* pTq, STqExecHandle* pExec, SSubmitReq* pReq, SMqDataR } if (pRsp->blockNum == 0) { - pRsp->skipLogNum++; return -1; } diff --git a/source/dnode/vnode/src/tq/tqPush.c b/source/dnode/vnode/src/tq/tqPush.c index e9e5f6cd8bc353c643546ba823f438ea9984b9e3..4b2160434fa8aab6a66ee3bed06a572c797aac50 100644 --- a/source/dnode/vnode/src/tq/tqPush.c +++ b/source/dnode/vnode/src/tq/tqPush.c @@ -242,6 +242,8 @@ int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver) void* data = taosMemoryMalloc(msgLen); if (data == NULL) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + qError("failed to copy data for stream since out of memory"); return -1; } memcpy(data, msg, msgLen); diff --git a/source/libs/stream/src/stream.c b/source/libs/stream/src/stream.c index 2a96db3bfc2c7f7440a288c3444ce10511e1f387..99a06575a948f19f42cb798fa3ea6dad4d4f7521 100644 --- a/source/libs/stream/src/stream.c +++ b/source/libs/stream/src/stream.c @@ -143,6 +143,9 @@ int32_t streamTaskEnqueueRetrieve(SStreamTask* pTask, SStreamRetrieveReq* pReq, // enqueue if (pData != NULL) { + qDebug("task %d(child %d) recv retrieve req from task %d, reqId %ld", pTask->taskId, pTask->selfChildId, + pReq->srcTaskId, pReq->reqId); + pData->type = STREAM_INPUT__DATA_RETRIEVE; pData->srcVgId = 0; // decode diff --git a/source/libs/stream/src/streamData.c b/source/libs/stream/src/streamData.c index dbf6350c93db32cd14a74e82434392aebe7b5dce..6be15222db244d5672c46fb36c7e1a9ef730c29c 100644 --- a/source/libs/stream/src/streamData.c +++ b/source/libs/stream/src/streamData.c @@ -57,7 +57,9 @@ int32_t streamRetrieveReqToData(const SStreamRetrieveReq* pReq, SStreamDataBlock pDataBlock->info.type = pRetrieve->streamBlockType; + pData->reqId = pReq->reqId; pData->blocks = pArray; + return 0; } diff --git a/source/libs/stream/src/streamDispatch.c b/source/libs/stream/src/streamDispatch.c index e2faf28abe2e0e6fa2e61abe7fbd9b1f00a8ff9e..5dec33d0fbfb4c22c1733bf29f2a7a61b9dfc183 100644 --- a/source/libs/stream/src/streamDispatch.c +++ b/source/libs/stream/src/streamDispatch.c @@ -65,6 +65,7 @@ int32_t tDecodeStreamDispatchReq(SDecoder* pDecoder, SStreamDispatchReq* pReq) { int32_t tEncodeStreamRetrieveReq(SEncoder* pEncoder, const SStreamRetrieveReq* pReq) { if (tStartEncode(pEncoder) < 0) return -1; if (tEncodeI64(pEncoder, pReq->streamId) < 0) return -1; + if (tEncodeI64(pEncoder, pReq->reqId) < 0) return -1; if (tEncodeI32(pEncoder, pReq->dstNodeId) < 0) return -1; if (tEncodeI32(pEncoder, pReq->dstTaskId) < 0) return -1; if (tEncodeI32(pEncoder, pReq->srcNodeId) < 0) return -1; @@ -77,6 +78,7 @@ int32_t tEncodeStreamRetrieveReq(SEncoder* pEncoder, const SStreamRetrieveReq* p int32_t tDecodeStreamRetrieveReq(SDecoder* pDecoder, SStreamRetrieveReq* pReq) { if (tStartDecode(pDecoder) < 0) return -1; if (tDecodeI64(pDecoder, &pReq->streamId) < 0) return -1; + if (tDecodeI64(pDecoder, &pReq->reqId) < 0) return -1; if (tDecodeI32(pDecoder, &pReq->dstNodeId) < 0) return -1; if (tDecodeI32(pDecoder, &pReq->dstTaskId) < 0) return -1; if (tDecodeI32(pDecoder, &pReq->srcNodeId) < 0) return -1; @@ -121,6 +123,7 @@ int32_t streamBroadcastToChildren(SStreamTask* pTask, const SSDataBlock* pBlock) int32_t sz = taosArrayGetSize(pTask->childEpInfo); ASSERT(sz > 0); for (int32_t i = 0; i < sz; i++) { + req.reqId = tGenIdPI64(); SStreamChildEpInfo* pEpInfo = taosArrayGetP(pTask->childEpInfo, i); req.dstNodeId = pEpInfo->nodeId; req.dstTaskId = pEpInfo->taskId; @@ -154,6 +157,9 @@ int32_t streamBroadcastToChildren(SStreamTask* pTask, const SSDataBlock* pBlock) ASSERT(0); return -1; } + + qDebug("task %d(child %d) send retrieve req to task %d at node %d, reqId %ld", pTask->taskId, pTask->selfChildId, + pEpInfo->taskId, pEpInfo->nodeId, req.reqId); } return 0; FAIL: diff --git a/source/libs/stream/src/streamExec.c b/source/libs/stream/src/streamExec.c index 9d22da86621d72e28748c94bef0dba84b4e431ba..f6eb9e32f28e18a75379ea0d789d9246f67fabe2 100644 --- a/source/libs/stream/src/streamExec.c +++ b/source/libs/stream/src/streamExec.c @@ -54,6 +54,9 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, void* data, SArray* pRes) block.info.type = STREAM_PULL_OVER; block.info.childId = pTask->selfChildId; taosArrayPush(pRes, &block); + + qDebug("task %d(child %d) processed retrieve, reqId %ld", pTask->taskId, pTask->selfChildId, + pRetrieveBlock->reqId); } break; } diff --git a/tests/test/c/tmqDemo.c b/tests/test/c/tmqDemo.c index 02fd3c13962eb5474deee298dca67dd3d07a7f49..61c50fb0e8e37effe7a89c1470e09c0e21fa14ce 100644 --- a/tests/test/c/tmqDemo.c +++ b/tests/test/c/tmqDemo.c @@ -388,13 +388,11 @@ void perf_loop(tmq_t* tmq, tmq_list_t* topics, int32_t totalMsgs, int64_t walLog } /*taosSsleep(3);*/ int32_t batchCnt = 0; - int32_t skipLogNum = 0; int64_t startTime = taosGetTimestampUs(); while (running) { TAOS_RES* tmqmessage = tmq_consumer_poll(tmq, 3000); if (tmqmessage) { batchCnt++; - /*skipLogNum += tmqGetSkipLogNum(tmqmessage);*/ if (0 != g_stConfInfo.showMsgFlag) { /*msg_process(tmqmessage);*/ } @@ -412,7 +410,7 @@ void perf_loop(tmq_t* tmq, tmq_list_t* topics, int32_t totalMsgs, int64_t walLog } if (0 == g_stConfInfo.simCase) { - printf("consume result: msgs: %d, skip log cnt: %d, time used:%.3f second\n", batchCnt, skipLogNum, consumeTime); + printf("consume result: msgs: %d, time used:%.3f second\n", batchCnt, consumeTime); } else { printf("{consume success: %d}", totalMsgs); }