Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
11e6856d
TDengine
项目概览
taosdata
/
TDengine
1 年多 前同步成功
通知
1185
Star
22016
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
提交
11e6856d
编写于
7月 06, 2023
作者:
wmmhello
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
fix:init char array
上级
a641a070
变更
3
显示空白变更内容
内联
并排
Showing
3 changed file
with
8 addition
and
8 deletion
+8
-8
source/client/src/clientTmq.c
source/client/src/clientTmq.c
+6
-6
source/dnode/vnode/src/tq/tq.c
source/dnode/vnode/src/tq/tq.c
+1
-1
source/dnode/vnode/src/tq/tqUtil.c
source/dnode/vnode/src/tq/tqUtil.c
+1
-1
未找到文件。
source/client/src/clientTmq.c
浏览文件 @
11e6856d
...
...
@@ -1414,7 +1414,7 @@ int32_t tmqPollCb(void* param, SDataBuf* pMsg, int32_t code) {
tDecoderClear
(
&
decoder
);
memcpy
(
&
pRspWrapper
->
dataRsp
,
pMsg
->
pData
,
sizeof
(
SMqRspHead
));
char
buf
[
TSDB_OFFSET_LEN
];
char
buf
[
TSDB_OFFSET_LEN
]
=
{
0
}
;
tFormatOffset
(
buf
,
TSDB_OFFSET_LEN
,
&
pRspWrapper
->
dataRsp
.
rspOffset
);
tscDebug
(
"consumer:0x%"
PRIx64
" recv poll rsp, vgId:%d, req ver:%"
PRId64
", rsp:%s type %d, reqId:0x%"
PRIx64
,
tmq
->
consumerId
,
vgId
,
pRspWrapper
->
dataRsp
.
reqOffset
.
version
,
buf
,
rspType
,
requestId
);
...
...
@@ -1559,7 +1559,7 @@ static bool doUpdateLocalEp(tmq_t* tmq, int32_t epoch, const SMqAskEpRsp* pRsp)
SMqClientVg
*
pVgCur
=
taosArrayGet
(
pTopicCur
->
vgs
,
j
);
makeTopicVgroupKey
(
vgKey
,
pTopicCur
->
topicName
,
pVgCur
->
vgId
);
char
buf
[
TSDB_OFFSET_LEN
];
char
buf
[
TSDB_OFFSET_LEN
]
=
{
0
}
;
tFormatOffset
(
buf
,
TSDB_OFFSET_LEN
,
&
pVgCur
->
offsetInfo
.
currentOffset
);
tscInfo
(
"consumer:0x%"
PRIx64
", epoch:%d vgId:%d vgKey:%s, offset:%s"
,
tmq
->
consumerId
,
epoch
,
pVgCur
->
vgId
,
vgKey
,
buf
);
...
...
@@ -1788,7 +1788,7 @@ static int32_t doTmqPollImpl(tmq_t* pTmq, SMqClientTopic* pTopic, SMqClientVg* p
sendInfo
->
msgType
=
TDMT_VND_TMQ_CONSUME
;
int64_t
transporterId
=
0
;
char
offsetFormatBuf
[
TSDB_OFFSET_LEN
];
char
offsetFormatBuf
[
TSDB_OFFSET_LEN
]
=
{
0
}
;
tFormatOffset
(
offsetFormatBuf
,
tListLen
(
offsetFormatBuf
),
&
pVg
->
offsetInfo
.
currentOffset
);
tscDebug
(
"consumer:0x%"
PRIx64
" send poll to %s vgId:%d, epoch %d, req:%s, reqId:0x%"
PRIx64
,
pTmq
->
consumerId
,
...
...
@@ -1925,7 +1925,7 @@ static void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) {
pVg
->
offsetInfo
.
walVerEnd
=
pDataRsp
->
head
.
walever
;
pVg
->
receivedInfoFromVnode
=
true
;
char
buf
[
TSDB_OFFSET_LEN
];
char
buf
[
TSDB_OFFSET_LEN
]
=
{
0
}
;
tFormatOffset
(
buf
,
TSDB_OFFSET_LEN
,
&
pDataRsp
->
rspOffset
);
if
(
pDataRsp
->
blockNum
==
0
)
{
tscDebug
(
"consumer:0x%"
PRIx64
" empty block received, vgId:%d, offset:%s, vg total:%"
PRId64
...
...
@@ -2033,7 +2033,7 @@ static void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) {
tmq
->
totalRows
+=
numOfRows
;
char
buf
[
TSDB_OFFSET_LEN
];
char
buf
[
TSDB_OFFSET_LEN
]
=
{
0
}
;
tFormatOffset
(
buf
,
TSDB_OFFSET_LEN
,
&
pVg
->
offsetInfo
.
currentOffset
);
tscDebug
(
"consumer:0x%"
PRIx64
" process taosx poll rsp, vgId:%d, offset:%s, blocks:%d, rows:%"
PRId64
", vg total:%"
PRId64
", total:%"
PRId64
", reqId:0x%"
PRIx64
,
...
...
@@ -2653,7 +2653,7 @@ int32_t tmq_get_topic_assignment(tmq_t* tmq, const char* pTopicName, tmq_topic_a
sendInfo
->
msgType
=
TDMT_VND_TMQ_VG_WALINFO
;
int64_t
transporterId
=
0
;
char
offsetFormatBuf
[
TSDB_OFFSET_LEN
];
char
offsetFormatBuf
[
TSDB_OFFSET_LEN
]
=
{
0
}
;
tFormatOffset
(
offsetFormatBuf
,
tListLen
(
offsetFormatBuf
),
&
pClientVg
->
offsetInfo
.
currentOffset
);
tscInfo
(
"consumer:0x%"
PRIx64
" %s retrieve wal info vgId:%d, epoch %d, req:%s, reqId:0x%"
PRIx64
,
...
...
source/dnode/vnode/src/tq/tq.c
浏览文件 @
11e6856d
...
...
@@ -510,7 +510,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
pHandle
->
epoch
=
reqEpoch
;
}
char
buf
[
TSDB_OFFSET_LEN
];
char
buf
[
TSDB_OFFSET_LEN
]
=
{
0
}
;
tFormatOffset
(
buf
,
TSDB_OFFSET_LEN
,
&
reqOffset
);
tqDebug
(
"tmq poll: consumer:0x%"
PRIx64
" (epoch %d), subkey %s, recv poll req vgId:%d, req:%s, reqId:0x%"
PRIx64
,
consumerId
,
req
.
epoch
,
pHandle
->
subKey
,
vgId
,
buf
,
req
.
reqId
);
...
...
source/dnode/vnode/src/tq/tqUtil.c
浏览文件 @
11e6856d
...
...
@@ -99,7 +99,7 @@ static int32_t extractResetOffsetVal(STqOffsetVal* pOffsetVal, STQ* pTq, STqHand
if
(
pOffset
!=
NULL
)
{
*
pOffsetVal
=
pOffset
->
val
;
char
formatBuf
[
TSDB_OFFSET_LEN
];
char
formatBuf
[
TSDB_OFFSET_LEN
]
=
{
0
}
;
tFormatOffset
(
formatBuf
,
TSDB_OFFSET_LEN
,
pOffsetVal
);
tqDebug
(
"tmq poll: consumer:0x%"
PRIx64
", subkey %s, vgId:%d, existed offset found, offset reset to %s and continue. reqId:0x%"
PRIx64
,
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录