xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

enh(tmq): enable background heartbeat by default

This commit is contained in:
Liu Jicong 2022-08-10 18:22:22 +08:00
parent 95f696b30d
commit 9e5d347383
3 changed files with 17 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
source/client/src/tmq.c
View File

@ -212,6 +212,7 @@ tmq_conf_t* tmq_conf_new() {
conf->autoCommit = true;
conf->autoCommitInterval = 5000;
conf->resetOffset = TMQ_CONF__RESET_OFFSET__EARLIEAST;
conf->hbBgEnable = true;
return conf;
}

18
source/dnode/mnode/impl/src/mndStream.c
View File

@ -929,14 +929,26 @@ static int32_t mndRetrieveStream(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pB
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)&pStream->status, true);
char sourceDB[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
tNameFromString(&n, pStream->sourceDb, T_NAME_ACCT | T_NAME_DB);
tNameGetDbName(&n, varDataVal(sourceDB));
varDataSetLen(sourceDB, strlen(varDataVal(sourceDB)));
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)&pStream->sourceDb, true);
colDataAppend(pColInfo, numOfRows, (const char *)&sourceDB, false);
char targetDB[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
tNameFromString(&n, pStream->targetDb, T_NAME_ACCT | T_NAME_DB);
tNameGetDbName(&n, varDataVal(targetDB));
varDataSetLen(targetDB, strlen(varDataVal(targetDB)));
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)&pStream->targetDb, true);
colDataAppend(pColInfo, numOfRows, (const char *)&targetDB, false);
char targetSTB[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
tNameFromString(&n, pStream->targetSTbName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
strcpy(&targetSTB[VARSTR_HEADER_SIZE], tNameGetTableName(&n));
varDataSetLen(targetSTB, strlen(varDataVal(targetSTB)));
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)&pStream->targetSTbName, true);
colDataAppend(pColInfo, numOfRows, (const char *)&targetSTB, false);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)&pStream->watermark, false);

2
source/util/src/terror.c
View File

@ -281,7 +281,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_TOPIC, "Invalid topic")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_TOPIC_QUERY, "Topic with invalid query")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_TOPIC_OPTION, "Topic with invalid option")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_CONSUMER_NOT_EXIST, "Consumer not exist")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOPIC_OPTION_UNCHNAGED, "Consumer unchanged")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOPIC_OPTION_UNCHNAGED, "Topic unchanged")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_SUBSCRIBE_NOT_EXIST, "Subcribe not exist")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_OFFSET_NOT_EXIST, "Offset not exist")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_CONSUMER_NOT_READY, "Consumer not ready")