Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
oceanbase
oblogclient
提交
aa5a0783
O
oblogclient
项目概览
oceanbase
/
oblogclient
1 年多 前同步成功
通知
5
Star
15
Fork
11
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
DevOps
流水线
流水线任务
计划
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
O
oblogclient
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
DevOps
DevOps
流水线
流水线任务
计划
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
流水线任务
提交
Issue看板
未验证
提交
aa5a0783
编写于
5月 19, 2022
作者:
H
He Wang
提交者:
GitHub
5月 19, 2022
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
format logger message (#39)
* format logger message * revert logic change
上级
1a92566f
变更
4
隐藏空白更改
内联
并排
Showing
4 changed file
with
20 addition
and
16 deletion
+20
-16
common/src/main/java/com/oceanbase/oms/logmessage/LogMessage.java
...rc/main/java/com/oceanbase/oms/logmessage/LogMessage.java
+2
-2
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/config/ObReaderConfig.java
...com/oceanbase/clogproxy/client/config/ObReaderConfig.java
+5
-1
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/connection/ClientHandler.java
.../oceanbase/clogproxy/client/connection/ClientHandler.java
+9
-9
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/connection/Connection.java
...com/oceanbase/clogproxy/client/connection/Connection.java
+4
-4
未找到文件。
common/src/main/java/com/oceanbase/oms/logmessage/LogMessage.java
浏览文件 @
aa5a0783
...
@@ -941,7 +941,7 @@ public class LogMessage extends DataMessage.Record {
...
@@ -941,7 +941,7 @@ public class LogMessage extends DataMessage.Record {
}
}
int
j
=
keyStr
.
indexOf
(
')'
,
i
);
int
j
=
keyStr
.
indexOf
(
')'
,
i
);
if
(
j
==
-
1
)
{
if
(
j
==
-
1
)
{
log
.
error
(
"
p
arse key error"
);
log
.
error
(
"
P
arse key error"
);
return
null
;
return
null
;
}
}
m
=
j
;
m
=
j
;
...
@@ -1077,7 +1077,7 @@ public class LogMessage extends DataMessage.Record {
...
@@ -1077,7 +1077,7 @@ public class LogMessage extends DataMessage.Record {
}
}
return
primaryValues
;
return
primaryValues
;
}
catch
(
Exception
e
)
{
}
catch
(
Exception
e
)
{
throw
new
RuntimeException
(
e
);
throw
new
LogMessageException
(
e
.
getMessage
(),
e
.
getCause
()
);
}
}
}
}
...
...
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/config/ObReaderConfig.java
浏览文件 @
aa5a0783
...
@@ -132,8 +132,12 @@ public class ObReaderConfig extends AbstractConnectionConfig {
...
@@ -132,8 +132,12 @@ public class ObReaderConfig extends AbstractConnectionConfig {
+
", cluster_password=******, "
+
", cluster_password=******, "
+
"tb_white_list="
+
"tb_white_list="
+
TABLE_WHITE_LIST
+
TABLE_WHITE_LIST
+
", tb_black_list="
+
TABLE_BLACK_LIST
+
", start_timestamp="
+
", start_timestamp="
+
START_TIMESTAMP
;
+
START_TIMESTAMP
+
", timezone="
+
TIME_ZONE
;
}
}
/**
/**
...
...
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/connection/ClientHandler.java
浏览文件 @
aa5a0783
...
@@ -226,7 +226,7 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
...
@@ -226,7 +226,7 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
byte
[]
bytes
=
new
byte
[
dataLength
];
byte
[]
bytes
=
new
byte
[
dataLength
];
buffer
.
readBytes
(
bytes
);
buffer
.
readBytes
(
bytes
);
LogProxyProto
.
RuntimeStatus
response
=
LogProxyProto
.
RuntimeStatus
.
parseFrom
(
bytes
);
LogProxyProto
.
RuntimeStatus
response
=
LogProxyProto
.
RuntimeStatus
.
parseFrom
(
bytes
);
logger
.
debug
(
"
s
erver status: {}"
,
response
.
toString
());
logger
.
debug
(
"
S
erver status: {}"
,
response
.
toString
());
state
=
HandshakeStateV1
.
PB_HEAD
;
state
=
HandshakeStateV1
.
PB_HEAD
;
}
else
{
}
else
{
dataNotEnough
=
true
;
dataNotEnough
=
true
;
...
@@ -252,18 +252,18 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
...
@@ -252,18 +252,18 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
*/
*/
private
void
checkHeader
(
int
version
,
int
type
,
int
length
)
{
private
void
checkHeader
(
int
version
,
int
type
,
int
length
)
{
if
(
ProtocolVersion
.
codeOf
(
version
)
==
null
)
{
if
(
ProtocolVersion
.
codeOf
(
version
)
==
null
)
{
logger
.
error
(
"
u
nsupported protocol version: {}"
,
version
);
logger
.
error
(
"
U
nsupported protocol version: {}"
,
version
);
throw
new
LogProxyClientException
(
throw
new
LogProxyClientException
(
ErrorCode
.
E_PROTOCOL
,
"
u
nsupported protocol version: "
+
version
);
ErrorCode
.
E_PROTOCOL
,
"
U
nsupported protocol version: "
+
version
);
}
}
if
(
HeaderType
.
codeOf
(
type
)
==
null
)
{
if
(
HeaderType
.
codeOf
(
type
)
==
null
)
{
logger
.
error
(
"
u
nsupported header type: {}"
,
type
);
logger
.
error
(
"
U
nsupported header type: {}"
,
type
);
throw
new
LogProxyClientException
(
throw
new
LogProxyClientException
(
ErrorCode
.
E_HEADER_TYPE
,
"
u
nsupported header type: "
+
type
);
ErrorCode
.
E_HEADER_TYPE
,
"
U
nsupported header type: "
+
type
);
}
}
if
(
length
<=
0
)
{
if
(
length
<=
0
)
{
logger
.
error
(
"
d
ata length equals 0"
);
logger
.
error
(
"
D
ata length equals 0"
);
throw
new
LogProxyClientException
(
ErrorCode
.
E_LEN
,
"
d
ata length equals 0"
);
throw
new
LogProxyClientException
(
ErrorCode
.
E_LEN
,
"
D
ata length equals 0"
);
}
}
}
}
...
@@ -453,8 +453,8 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
...
@@ -453,8 +453,8 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
poolFlag
=
false
;
poolFlag
=
false
;
logger
.
info
(
logger
.
info
(
"C
onnect broken of ClientId: {} with
LogProxy: {}"
,
"C
hannel closed with ClientId: {},
LogProxy: {}"
,
params
.
info
(),
params
.
getClientId
(),
NetworkUtil
.
parseRemoteAddress
(
ctx
.
channel
()));
NetworkUtil
.
parseRemoteAddress
(
ctx
.
channel
()));
ctx
.
channel
().
disconnect
();
ctx
.
channel
().
disconnect
();
ctx
.
close
();
ctx
.
close
();
...
...
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/connection/Connection.java
浏览文件 @
aa5a0783
...
@@ -41,7 +41,7 @@ public class Connection {
...
@@ -41,7 +41,7 @@ public class Connection {
/** Close this connection. */
/** Close this connection. */
public
void
close
()
{
public
void
close
()
{
if
(!
closed
.
compareAndSet
(
false
,
true
))
{
if
(!
closed
.
compareAndSet
(
false
,
true
))
{
logger
.
warn
(
"
c
onnection already closed"
);
logger
.
warn
(
"
C
onnection already closed"
);
}
}
if
(
channel
!=
null
)
{
if
(
channel
!=
null
)
{
if
(
channel
.
isActive
())
{
if
(
channel
.
isActive
())
{
...
@@ -49,7 +49,7 @@ public class Connection {
...
@@ -49,7 +49,7 @@ public class Connection {
channel
.
close
().
addListener
(
this
::
logCloseResult
).
syncUninterruptibly
();
channel
.
close
().
addListener
(
this
::
logCloseResult
).
syncUninterruptibly
();
}
catch
(
Exception
e
)
{
}
catch
(
Exception
e
)
{
logger
.
warn
(
logger
.
warn
(
"
c
lose connection to remote address {} exception"
,
"
C
lose connection to remote address {} exception"
,
NetworkUtil
.
parseRemoteAddress
(
channel
),
NetworkUtil
.
parseRemoteAddress
(
channel
),
e
);
e
);
}
}
...
@@ -68,12 +68,12 @@ public class Connection {
...
@@ -68,12 +68,12 @@ public class Connection {
if
(
future
.
isSuccess
())
{
if
(
future
.
isSuccess
())
{
if
(
logger
.
isInfoEnabled
())
{
if
(
logger
.
isInfoEnabled
())
{
logger
.
info
(
logger
.
info
(
"
c
lose connection to remote address {} success"
,
"
C
lose connection to remote address {} success"
,
NetworkUtil
.
parseRemoteAddress
(
channel
));
NetworkUtil
.
parseRemoteAddress
(
channel
));
}
}
}
else
{
}
else
{
logger
.
warn
(
logger
.
warn
(
"
c
lose connection to remote address {} fail"
,
"
C
lose connection to remote address {} fail"
,
NetworkUtil
.
parseRemoteAddress
(
channel
),
NetworkUtil
.
parseRemoteAddress
(
channel
),
future
.
cause
());
future
.
cause
());
}
}
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录