Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
e921117a
T
TDengine
项目概览
taosdata
/
TDengine
大约 1 年 前同步成功
通知
1184
Star
22015
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
T
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
未验证
提交
e921117a
编写于
7月 07, 2023
作者:
H
Haojun Liao
提交者:
GitHub
7月 07, 2023
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #22000 from taosdata/mark/tmq
fix:set commitOffset is send msg success & optimize log & add test ca…
上级
b7bf4925
4436eb7e
变更
3
隐藏空白更改
内联
并排
Showing
3 changed file
with
145 addition
and
3 deletion
+145
-3
source/client/src/clientTmq.c
source/client/src/clientTmq.c
+3
-1
source/client/test/clientTests.cpp
source/client/test/clientTests.cpp
+140
-0
source/dnode/vnode/src/tq/tq.c
source/dnode/vnode/src/tq/tq.c
+2
-2
未找到文件。
source/client/src/clientTmq.c
浏览文件 @
e921117a
...
...
@@ -678,6 +678,8 @@ static void asyncCommitOffset(tmq_t* tmq, const TAOS_RES* pRes, int32_t type, tm
taosMemoryFree
(
pParamSet
);
pCommitFp
(
tmq
,
code
,
userParam
);
}
// update the offset value.
pVg
->
offsetInfo
.
committedOffset
=
pVg
->
offsetInfo
.
currentOffset
;
}
else
{
// do not perform commit, callback user function directly.
taosMemoryFree
(
pParamSet
);
pCommitFp
(
tmq
,
code
,
userParam
);
...
...
@@ -2712,7 +2714,7 @@ int32_t tmq_get_topic_assignment(tmq_t* tmq, const char* pTopicName, tmq_topic_a
// char offsetBuf[TSDB_OFFSET_LEN] = {0};
// tFormatOffset(offsetBuf, tListLen(offsetBuf), &pOffsetInfo->currentOffset);
tscInfo
(
"vgId:%d offset is
old
to:%"
PRId64
,
p
->
vgId
,
p
->
currentOffset
);
tscInfo
(
"vgId:%d offset is
update
to:%"
PRId64
,
p
->
vgId
,
p
->
currentOffset
);
pOffsetInfo
->
walVerBegin
=
p
->
begin
;
pOffsetInfo
->
walVerEnd
=
p
->
end
;
...
...
source/client/test/clientTests.cpp
浏览文件 @
e921117a
...
...
@@ -1073,6 +1073,146 @@ TEST(clientCase, sub_db_test) {
fprintf
(
stderr
,
"%d msg consumed, include %d rows
\n
"
,
msgCnt
,
totalRows
);
}
TEST
(
clientCase
,
td_25129
)
{
// taos_options(TSDB_OPTION_CONFIGDIR, "~/first/cfg");
TAOS
*
pConn
=
taos_connect
(
"localhost"
,
"root"
,
"taosdata"
,
NULL
,
0
);
ASSERT_NE
(
pConn
,
nullptr
);
tmq_conf_t
*
conf
=
tmq_conf_new
();
tmq_conf_set
(
conf
,
"enable.auto.commit"
,
"false"
);
tmq_conf_set
(
conf
,
"auto.commit.interval.ms"
,
"2000"
);
tmq_conf_set
(
conf
,
"group.id"
,
"group_id_2"
);
tmq_conf_set
(
conf
,
"td.connect.user"
,
"root"
);
tmq_conf_set
(
conf
,
"td.connect.pass"
,
"taosdata"
);
tmq_conf_set
(
conf
,
"auto.offset.reset"
,
"earliest"
);
tmq_conf_set
(
conf
,
"msg.with.table.name"
,
"true"
);
tmq_t
*
tmq
=
tmq_consumer_new
(
conf
,
NULL
,
0
);
tmq_conf_destroy
(
conf
);
// 创建订阅 topics 列表
tmq_list_t
*
topicList
=
tmq_list_new
();
tmq_list_append
(
topicList
,
"tp"
);
// 启动订阅
tmq_subscribe
(
tmq
,
topicList
);
tmq_list_destroy
(
topicList
);
TAOS_FIELD
*
fields
=
NULL
;
int32_t
numOfFields
=
0
;
int32_t
precision
=
0
;
int32_t
totalRows
=
0
;
int32_t
msgCnt
=
0
;
int32_t
timeout
=
2000
;
int32_t
count
=
0
;
tmq_topic_assignment
*
pAssign
=
NULL
;
int32_t
numOfAssign
=
0
;
int32_t
code
=
tmq_get_topic_assignment
(
tmq
,
"tp"
,
&
pAssign
,
&
numOfAssign
);
if
(
code
!=
0
)
{
printf
(
"error occurs:%s
\n
"
,
tmq_err2str
(
code
));
tmq_free_assignment
(
pAssign
);
tmq_consumer_close
(
tmq
);
taos_close
(
pConn
);
fprintf
(
stderr
,
"%d msg consumed, include %d rows
\n
"
,
msgCnt
,
totalRows
);
return
;
}
for
(
int
i
=
0
;
i
<
numOfAssign
;
i
++
){
printf
(
"assign i:%d, vgId:%d, offset:%lld, start:%lld, end:%lld
\n
"
,
i
,
pAssign
[
i
].
vgId
,
pAssign
[
i
].
currentOffset
,
pAssign
[
i
].
begin
,
pAssign
[
i
].
end
);
}
// tmq_offset_seek(tmq, "tp", pAssign[0].vgId, 4);
tmq_free_assignment
(
pAssign
);
code
=
tmq_get_topic_assignment
(
tmq
,
"tp"
,
&
pAssign
,
&
numOfAssign
);
if
(
code
!=
0
)
{
printf
(
"error occurs:%s
\n
"
,
tmq_err2str
(
code
));
tmq_free_assignment
(
pAssign
);
tmq_consumer_close
(
tmq
);
taos_close
(
pConn
);
fprintf
(
stderr
,
"%d msg consumed, include %d rows
\n
"
,
msgCnt
,
totalRows
);
return
;
}
for
(
int
i
=
0
;
i
<
numOfAssign
;
i
++
){
printf
(
"assign i:%d, vgId:%d, offset:%lld, start:%lld, end:%lld
\n
"
,
i
,
pAssign
[
i
].
vgId
,
pAssign
[
i
].
currentOffset
,
pAssign
[
i
].
begin
,
pAssign
[
i
].
end
);
}
tmq_free_assignment
(
pAssign
);
code
=
tmq_get_topic_assignment
(
tmq
,
"tp"
,
&
pAssign
,
&
numOfAssign
);
if
(
code
!=
0
)
{
printf
(
"error occurs:%s
\n
"
,
tmq_err2str
(
code
));
tmq_free_assignment
(
pAssign
);
tmq_consumer_close
(
tmq
);
taos_close
(
pConn
);
fprintf
(
stderr
,
"%d msg consumed, include %d rows
\n
"
,
msgCnt
,
totalRows
);
return
;
}
for
(
int
i
=
0
;
i
<
numOfAssign
;
i
++
){
printf
(
"assign i:%d, vgId:%d, offset:%lld, start:%lld, end:%lld
\n
"
,
i
,
pAssign
[
i
].
vgId
,
pAssign
[
i
].
currentOffset
,
pAssign
[
i
].
begin
,
pAssign
[
i
].
end
);
}
while
(
1
)
{
TAOS_RES
*
pRes
=
tmq_consumer_poll
(
tmq
,
timeout
);
if
(
pRes
)
{
char
buf
[
128
];
const
char
*
topicName
=
tmq_get_topic_name
(
pRes
);
// const char* dbName = tmq_get_db_name(pRes);
// int32_t vgroupId = tmq_get_vgroup_id(pRes);
//
// printf("topic: %s\n", topicName);
// printf("db: %s\n", dbName);
// printf("vgroup id: %d\n", vgroupId);
printSubResults
(
pRes
,
&
totalRows
);
}
else
{
tmq_offset_seek
(
tmq
,
"tp"
,
pAssign
[
0
].
vgId
,
pAssign
[
0
].
currentOffset
);
tmq_offset_seek
(
tmq
,
"tp"
,
pAssign
[
1
].
vgId
,
pAssign
[
1
].
currentOffset
);
continue
;
}
// tmq_commit_sync(tmq, pRes);
if
(
pRes
!=
NULL
)
{
taos_free_result
(
pRes
);
// if ((++count) > 1) {
// break;
// }
}
else
{
break
;
}
// tmq_offset_seek(tmq, "tp", pAssign[0].vgId, pAssign[0].begin);
}
tmq_free_assignment
(
pAssign
);
code
=
tmq_get_topic_assignment
(
tmq
,
"tp"
,
&
pAssign
,
&
numOfAssign
);
if
(
code
!=
0
)
{
printf
(
"error occurs:%s
\n
"
,
tmq_err2str
(
code
));
tmq_free_assignment
(
pAssign
);
tmq_consumer_close
(
tmq
);
taos_close
(
pConn
);
fprintf
(
stderr
,
"%d msg consumed, include %d rows
\n
"
,
msgCnt
,
totalRows
);
return
;
}
for
(
int
i
=
0
;
i
<
numOfAssign
;
i
++
){
printf
(
"assign i:%d, vgId:%d, offset:%lld, start:%lld, end:%lld
\n
"
,
i
,
pAssign
[
i
].
vgId
,
pAssign
[
i
].
currentOffset
,
pAssign
[
i
].
begin
,
pAssign
[
i
].
end
);
}
tmq_consumer_close
(
tmq
);
taos_close
(
pConn
);
fprintf
(
stderr
,
"%d msg consumed, include %d rows
\n
"
,
msgCnt
,
totalRows
);
}
TEST
(
clientCase
,
sub_tb_test
)
{
taos_options
(
TSDB_OPTION_CONFIGDIR
,
"~/first/cfg"
);
...
...
source/dnode/vnode/src/tq/tq.c
浏览文件 @
e921117a
...
...
@@ -490,7 +490,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
if
(
!
exec
)
{
tqSetHandleExec
(
pHandle
);
// qSetTaskCode(pHandle->execHandle.task, TDB_CODE_SUCCESS);
tqDebug
(
"tmq poll: consumer:0x%"
PRIx64
"vgId:%d, topic:%s, set handle exec, pHandle:%p"
,
consumerId
,
vgId
,
tqDebug
(
"tmq poll: consumer:0x%"
PRIx64
"
vgId:%d, topic:%s, set handle exec, pHandle:%p"
,
consumerId
,
vgId
,
req
.
subKey
,
pHandle
);
taosWUnLockLatch
(
&
pTq
->
lock
);
break
;
...
...
@@ -518,7 +518,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
code
=
tqExtractDataForMq
(
pTq
,
pHandle
,
&
req
,
pMsg
);
tqSetHandleIdle
(
pHandle
);
tqDebug
(
"tmq poll: consumer:0x%"
PRIx64
"
vgId:%d, topic:%s,
, set handle idle, pHandle:%p"
,
consumerId
,
vgId
,
tqDebug
(
"tmq poll: consumer:0x%"
PRIx64
"
vgId:%d, topic:%s
, set handle idle, pHandle:%p"
,
consumerId
,
vgId
,
req
.
subKey
,
pHandle
);
return
code
;
}
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录