diff --git a/tests/test/c/tmqSim.c b/tests/test/c/tmqSim.c index 59005fae945d863cbe28bc0b5e994abea83405d3..c2fba52e6eb301fa5b528e37950084f164501efe 100644 --- a/tests/test/c/tmqSim.c +++ b/tests/test/c/tmqSim.c @@ -22,9 +22,9 @@ #include #include "taos.h" +#include "taosdef.h" #include "taoserror.h" #include "tlog.h" -#include "taosdef.h" #include "types.h" #define GREEN "\033[1;32m" @@ -36,11 +36,7 @@ #define MAX_CONSUMER_THREAD_CNT (16) #define MAX_VGROUP_CNT (32) -typedef enum { - NOTIFY_CMD_START_CONSUM, - NOTIFY_CMD_START_COMMIT, - NOTIFY_CMD_ID_BUTT -}NOTIFY_CMD_ID; +typedef enum { NOTIFY_CMD_START_CONSUM, NOTIFY_CMD_START_COMMIT, NOTIFY_CMD_ID_BUTT } NOTIFY_CMD_ID; typedef struct { TdThread thread; @@ -52,8 +48,8 @@ typedef struct { // char autoOffsetRest[16]; // none, earliest, latest TdFilePtr pConsumeRowsFile; - int32_t ifCheckData; - int64_t expectMsgCnt; + int32_t ifCheckData; + int64_t expectMsgCnt; int64_t consumeMsgCnt; int64_t consumeRowCnt; @@ -89,6 +85,7 @@ typedef struct { int32_t saveRowFlag; int32_t consumeDelay; // unit s int32_t numOfThread; + int32_t useSnapshot; SThreadInfo stThreads[MAX_CONSUMER_THREAD_CNT]; } SConfInfo; @@ -217,6 +214,8 @@ void parseArgument(int32_t argc, char* argv[]) { g_stConfInfo.saveRowFlag = atol(argv[++i]); } else if (strcmp(argv[i], "-y") == 0) { g_stConfInfo.consumeDelay = atol(argv[++i]); + } else if (strcmp(argv[i], "-e") == 0) { + g_stConfInfo.useSnapshot = atol(argv[++i]); } else { pError("%s unknow para: %s %s", GREEN, argv[++i], NC); exit(-1); @@ -310,11 +309,11 @@ int32_t saveConsumeContentToTbl(SThreadInfo* pInfo, char* buf) { return 0; } -static char *shellFormatTimestamp(char *buf, int64_t val, int32_t precision) { - //if (shell.args.is_raw_time) { - // sprintf(buf, "%" PRId64, val); - // return buf; - //} +static char* shellFormatTimestamp(char* buf, int64_t val, int32_t precision) { + // if (shell.args.is_raw_time) { + // sprintf(buf, "%" PRId64, val); + // return buf; + // } time_t tt; int32_t ms = 0; @@ -352,7 +351,7 @@ static char *shellFormatTimestamp(char *buf, int64_t val, int32_t precision) { } } - struct tm *ptm = taosLocalTime(&tt, NULL); + struct tm* ptm = taosLocalTime(&tt, NULL); size_t pos = strftime(buf, 35, "%Y-%m-%d %H:%M:%S", ptm); if (precision == TSDB_TIME_PRECISION_NANO) { @@ -366,7 +365,8 @@ static char *shellFormatTimestamp(char *buf, int64_t val, int32_t precision) { return buf; } -static void shellDumpFieldToFile(TdFilePtr pFile, const char *val, TAOS_FIELD *field, int32_t length, int32_t precision) { +static void shellDumpFieldToFile(TdFilePtr pFile, const char* val, TAOS_FIELD* field, int32_t length, + int32_t precision) { if (val == NULL) { taosFprintfFile(pFile, "%s", TSDB_DATA_NULL_STR); return; @@ -376,31 +376,31 @@ static void shellDumpFieldToFile(TdFilePtr pFile, const char *val, TAOS_FIELD *f char buf[TSDB_MAX_BYTES_PER_ROW]; switch (field->type) { case TSDB_DATA_TYPE_BOOL: - taosFprintfFile(pFile, "%d", ((((int32_t)(*((char *)val))) == 1) ? 1 : 0)); + taosFprintfFile(pFile, "%d", ((((int32_t)(*((char*)val))) == 1) ? 1 : 0)); break; case TSDB_DATA_TYPE_TINYINT: - taosFprintfFile(pFile, "%d", *((int8_t *)val)); + taosFprintfFile(pFile, "%d", *((int8_t*)val)); break; case TSDB_DATA_TYPE_UTINYINT: - taosFprintfFile(pFile, "%u", *((uint8_t *)val)); + taosFprintfFile(pFile, "%u", *((uint8_t*)val)); break; case TSDB_DATA_TYPE_SMALLINT: - taosFprintfFile(pFile, "%d", *((int16_t *)val)); + taosFprintfFile(pFile, "%d", *((int16_t*)val)); break; case TSDB_DATA_TYPE_USMALLINT: - taosFprintfFile(pFile, "%u", *((uint16_t *)val)); + taosFprintfFile(pFile, "%u", *((uint16_t*)val)); break; case TSDB_DATA_TYPE_INT: - taosFprintfFile(pFile, "%d", *((int32_t *)val)); + taosFprintfFile(pFile, "%d", *((int32_t*)val)); break; case TSDB_DATA_TYPE_UINT: - taosFprintfFile(pFile, "%u", *((uint32_t *)val)); + taosFprintfFile(pFile, "%u", *((uint32_t*)val)); break; case TSDB_DATA_TYPE_BIGINT: - taosFprintfFile(pFile, "%" PRId64, *((int64_t *)val)); + taosFprintfFile(pFile, "%" PRId64, *((int64_t*)val)); break; case TSDB_DATA_TYPE_UBIGINT: - taosFprintfFile(pFile, "%" PRIu64, *((uint64_t *)val)); + taosFprintfFile(pFile, "%" PRIu64, *((uint64_t*)val)); break; case TSDB_DATA_TYPE_FLOAT: taosFprintfFile(pFile, "%.5f", GET_FLOAT_VAL(val)); @@ -421,7 +421,7 @@ static void shellDumpFieldToFile(TdFilePtr pFile, const char *val, TAOS_FIELD *f taosFprintfFile(pFile, "\'%s\'", buf); break; case TSDB_DATA_TYPE_TIMESTAMP: - shellFormatTimestamp(buf, *(int64_t *)val, precision); + shellFormatTimestamp(buf, *(int64_t*)val, precision); taosFprintfFile(pFile, "'%s'", buf); break; default: @@ -429,12 +429,13 @@ static void shellDumpFieldToFile(TdFilePtr pFile, const char *val, TAOS_FIELD *f } } -static void dumpToFileForCheck(TdFilePtr pFile, TAOS_ROW row, TAOS_FIELD* fields, int32_t* length, int32_t num_fields, int32_t precision) { +static void dumpToFileForCheck(TdFilePtr pFile, TAOS_ROW row, TAOS_FIELD* fields, int32_t* length, int32_t num_fields, + int32_t precision) { for (int32_t i = 0; i < num_fields; i++) { if (i > 0) { taosFprintfFile(pFile, "\n"); } - shellDumpFieldToFile(pFile, (const char *)row[i], fields + i, length[i], precision); + shellDumpFieldToFile(pFile, (const char*)row[i], fields + i, length[i], precision); } taosFprintfFile(pFile, "\n"); } @@ -444,40 +445,42 @@ static int32_t msg_process(TAOS_RES* msg, SThreadInfo* pInfo, int32_t msgIndex) int32_t totalRows = 0; // printf("topic: %s\n", tmq_get_topic_name(msg)); - int32_t vgroupId = tmq_get_vgroup_id(msg); - const char* dbName = tmq_get_db_name(msg); + int32_t vgroupId = tmq_get_vgroup_id(msg); + const char* dbName = tmq_get_db_name(msg); taosFprintfFile(g_fp, "consumerId: %d, msg index:%" PRId64 "\n", pInfo->consumerId, msgIndex); - taosFprintfFile(g_fp, "dbName: %s, topic: %s, vgroupId: %d\n", dbName != NULL ? dbName : "invalid table", tmq_get_topic_name(msg), vgroupId); + taosFprintfFile(g_fp, "dbName: %s, topic: %s, vgroupId: %d\n", dbName != NULL ? dbName : "invalid table", + tmq_get_topic_name(msg), vgroupId); while (1) { TAOS_ROW row = taos_fetch_row(msg); if (row == NULL) break; - TAOS_FIELD* fields = taos_fetch_fields(msg); + TAOS_FIELD* fields = taos_fetch_fields(msg); int32_t numOfFields = taos_field_count(msg); - int32_t* length = taos_fetch_lengths(msg); - int32_t precision = taos_result_precision(msg); - const char* tbName = tmq_get_table_name(msg); + int32_t* length = taos_fetch_lengths(msg); + int32_t precision = taos_result_precision(msg); + const char* tbName = tmq_get_table_name(msg); - #if 0 + #if 0 // get schema //============================== stub =================================================// for (int32_t i = 0; i < numOfFields; i++) { taosFprintfFile(g_fp, "%02d: name: %s, type: %d, len: %d\n", i, fields[i].name, fields[i].type, fields[i].bytes); } //============================== stub =================================================// - #endif - - dumpToFileForCheck(pInfo->pConsumeRowsFile, row, fields, length, numOfFields, precision); + #endif + + dumpToFileForCheck(pInfo->pConsumeRowsFile, row, fields, length, numOfFields, precision); + taos_print_row(buf, row, fields, numOfFields); if (0 != g_stConfInfo.showRowFlag) { taosFprintfFile(g_fp, "tbname:%s, rows[%d]: %s\n", (tbName != NULL ? tbName : "null table"), totalRows, buf); - //if (0 != g_stConfInfo.saveRowFlag) { - // saveConsumeContentToTbl(pInfo, buf); - //} + // if (0 != g_stConfInfo.saveRowFlag) { + // saveConsumeContentToTbl(pInfo, buf); + // } } totalRows++; @@ -500,8 +503,7 @@ int queryDB(TAOS* taos, char* command) { return 0; } -static void appNothing(void* param, TAOS_RES* res, int32_t numOfRows) { -} +static void appNothing(void* param, TAOS_RES* res, int32_t numOfRows) {} int32_t notifyMainScript(SThreadInfo* pInfo, int32_t cmdId) { char sqlStr[1024] = {0}; @@ -509,11 +511,8 @@ int32_t notifyMainScript(SThreadInfo* pInfo, int32_t cmdId) { int64_t now = taosGetTimestampMs(); // schema: ts timestamp, consumerid int, consummsgcnt bigint, checkresult int - sprintf(sqlStr, "insert into %s.notifyinfo values (%"PRId64", %d, %d)", - g_stConfInfo.cdbName, - now, - cmdId, - pInfo->consumerId); + sprintf(sqlStr, "insert into %s.notifyinfo values (%" PRId64 ", %d, %d)", g_stConfInfo.cdbName, now, cmdId, + pInfo->consumerId); taos_query_a(pInfo->taos, sqlStr, appNothing, NULL); @@ -523,12 +522,12 @@ int32_t notifyMainScript(SThreadInfo* pInfo, int32_t cmdId) { } static int32_t g_once_commit_flag = 0; -static void tmq_commit_cb_print(tmq_t* tmq, int32_t code, void* param) { - pError("tmq_commit_cb_print() commit %d\n", code); +static void tmq_commit_cb_print(tmq_t* tmq, int32_t code, void* param) { + pError("tmq_commit_cb_print() commit %d\n", code); if (0 == g_once_commit_flag) { g_once_commit_flag = 1; - notifyMainScript((SThreadInfo*)param, (int32_t)NOTIFY_CMD_START_COMMIT); + notifyMainScript((SThreadInfo*)param, (int32_t)NOTIFY_CMD_START_COMMIT); } char tmpString[128]; @@ -564,6 +563,10 @@ void build_consumer(SThreadInfo* pInfo) { // tmq_conf_set(conf, "auto.offset.reset", "none"); // tmq_conf_set(conf, "auto.offset.reset", "earliest"); // tmq_conf_set(conf, "auto.offset.reset", "latest"); + // + if (g_stConfInfo.useSnapshot) { + tmq_conf_set(conf, "experiment.use.snapshot", "true"); + } pInfo->tmq = tmq_consumer_new(conf, NULL, 0); @@ -618,12 +621,13 @@ void loop_consume(SThreadInfo* pInfo) { pInfo->consumerId); pInfo->ts = taosGetTimestampMs(); - + if (pInfo->ifCheckData) { - char filename[256] = {0}; + char filename[256] = {0}; char tmpString[128]; - //sprintf(filename, "%s/../log/consumerid_%d_%s.txt", configDir, pInfo->consumerId, getCurrentTimeString(tmpString)); - sprintf(filename, "%s/../log/consumerid_%d.txt", configDir, pInfo->consumerId); + // sprintf(filename, "%s/../log/consumerid_%d_%s.txt", configDir, pInfo->consumerId, + // getCurrentTimeString(tmpString)); + sprintf(filename, "%s/../log/consumerid_%d.txt", configDir, pInfo->consumerId); pInfo->pConsumeRowsFile = taosOpenFile(filename, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_STREAM); if (pInfo->pConsumeRowsFile == NULL) { taosFprintfFile(g_fp, "%s create file fail for save rows content\n", getCurrentTimeString(tmpString)); @@ -642,10 +646,10 @@ void loop_consume(SThreadInfo* pInfo) { totalMsgs++; - if (0 == once_flag) { + if (0 == once_flag) { once_flag = 1; - notifyMainScript(pInfo, NOTIFY_CMD_START_CONSUM); - } + notifyMainScript(pInfo, NOTIFY_CMD_START_CONSUM); + } if (totalRows >= pInfo->expectMsgCnt) { char tmpString[128]; @@ -678,7 +682,7 @@ void* consumeThreadFunc(void* param) { pInfo->taos = taos_connect(NULL, "root", "taosdata", NULL, 0); if (pInfo->taos == NULL) { taosFprintfFile(g_fp, "taos_connect() fail, can not notify and save consume result to main scripte\n"); - return NULL; + return NULL; } build_consumer(pInfo);