Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
DiDi
kafka-manager
提交
949d6ba6
K
kafka-manager
项目概览
DiDi
/
kafka-manager
10 个月 前同步成功
通知
58
Star
6372
Fork
1229
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
K
kafka-manager
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
前往新版Gitcode,体验更适合开发者的 AI 搜索 >>
提交
949d6ba6
编写于
9月 06, 2022
作者:
Z
zengqiao
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
集群Broker列表,增加Controller角色信息
上级
ceb8db09
变更
2
隐藏空白更改
内联
并排
Showing
2 changed file
with
15 addition
and
4 deletion
+15
-4
km-biz/src/main/java/com/xiaojukeji/know/streaming/km/biz/cluster/impl/ClusterBrokersManagerImpl.java
...eaming/km/biz/cluster/impl/ClusterBrokersManagerImpl.java
+13
-4
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/constant/KafkaConstant.java
...keji/know/streaming/km/common/constant/KafkaConstant.java
+2
-0
未找到文件。
km-biz/src/main/java/com/xiaojukeji/know/streaming/km/biz/cluster/impl/ClusterBrokersManagerImpl.java
浏览文件 @
949d6ba6
...
...
@@ -14,6 +14,7 @@ import com.xiaojukeji.know.streaming.km.common.bean.entity.topic.Topic;
import
com.xiaojukeji.know.streaming.km.common.bean.vo.cluster.res.ClusterBrokersOverviewVO
;
import
com.xiaojukeji.know.streaming.km.common.bean.vo.cluster.res.ClusterBrokersStateVO
;
import
com.xiaojukeji.know.streaming.km.common.bean.vo.kafkacontroller.KafkaControllerVO
;
import
com.xiaojukeji.know.streaming.km.common.constant.KafkaConstant
;
import
com.xiaojukeji.know.streaming.km.common.enums.SortTypeEnum
;
import
com.xiaojukeji.know.streaming.km.common.utils.PaginationMetricsUtil
;
import
com.xiaojukeji.know.streaming.km.common.utils.PaginationUtil
;
...
...
@@ -71,6 +72,9 @@ public class ClusterBrokersManagerImpl implements ClusterBrokersManager {
Topic
groupTopic
=
topicService
.
getTopic
(
clusterPhyId
,
org
.
apache
.
kafka
.
common
.
internals
.
Topic
.
GROUP_METADATA_TOPIC_NAME
);
Topic
transactionTopic
=
topicService
.
getTopic
(
clusterPhyId
,
org
.
apache
.
kafka
.
common
.
internals
.
Topic
.
TRANSACTION_STATE_TOPIC_NAME
);
//获取controller信息
KafkaController
kafkaController
=
kafkaControllerService
.
getKafkaControllerFromDB
(
clusterPhyId
);
// 格式转换
return
PaginationResult
.
buildSuc
(
this
.
convert2ClusterBrokersOverviewVOList
(
...
...
@@ -78,7 +82,8 @@ public class ClusterBrokersManagerImpl implements ClusterBrokersManager {
brokerList
,
metricsResult
.
getData
(),
groupTopic
,
transactionTopic
transactionTopic
,
kafkaController
),
paginationResult
);
...
...
@@ -159,7 +164,8 @@ public class ClusterBrokersManagerImpl implements ClusterBrokersManager {
List
<
Broker
>
brokerList
,
List
<
BrokerMetrics
>
metricsList
,
Topic
groupTopic
,
Topic
transactionTopic
)
{
Topic
transactionTopic
,
KafkaController
kafkaController
)
{
Map
<
Integer
,
BrokerMetrics
>
metricsMap
=
metricsList
==
null
?
new
HashMap
<>():
metricsList
.
stream
().
collect
(
Collectors
.
toMap
(
BrokerMetrics:
:
getBrokerId
,
Function
.
identity
()));
Map
<
Integer
,
Broker
>
brokerMap
=
brokerList
==
null
?
new
HashMap
<>():
brokerList
.
stream
().
collect
(
Collectors
.
toMap
(
Broker:
:
getBrokerId
,
Function
.
identity
()));
...
...
@@ -169,12 +175,12 @@ public class ClusterBrokersManagerImpl implements ClusterBrokersManager {
Broker
broker
=
brokerMap
.
get
(
brokerId
);
BrokerMetrics
brokerMetrics
=
metricsMap
.
get
(
brokerId
);
voList
.
add
(
this
.
convert2ClusterBrokersOverviewVO
(
brokerId
,
broker
,
brokerMetrics
,
groupTopic
,
transactionTopic
));
voList
.
add
(
this
.
convert2ClusterBrokersOverviewVO
(
brokerId
,
broker
,
brokerMetrics
,
groupTopic
,
transactionTopic
,
kafkaController
));
}
return
voList
;
}
private
ClusterBrokersOverviewVO
convert2ClusterBrokersOverviewVO
(
Integer
brokerId
,
Broker
broker
,
BrokerMetrics
brokerMetrics
,
Topic
groupTopic
,
Topic
transactionTopic
)
{
private
ClusterBrokersOverviewVO
convert2ClusterBrokersOverviewVO
(
Integer
brokerId
,
Broker
broker
,
BrokerMetrics
brokerMetrics
,
Topic
groupTopic
,
Topic
transactionTopic
,
KafkaController
kafkaController
)
{
ClusterBrokersOverviewVO
clusterBrokersOverviewVO
=
new
ClusterBrokersOverviewVO
();
clusterBrokersOverviewVO
.
setBrokerId
(
brokerId
);
if
(
broker
!=
null
)
{
...
...
@@ -192,6 +198,9 @@ public class ClusterBrokersManagerImpl implements ClusterBrokersManager {
if
(
transactionTopic
!=
null
&&
transactionTopic
.
getBrokerIdSet
().
contains
(
brokerId
))
{
clusterBrokersOverviewVO
.
getKafkaRoleList
().
add
(
transactionTopic
.
getTopicName
());
}
if
(
kafkaController
!=
null
&&
kafkaController
.
getBrokerId
().
equals
(
brokerId
))
{
clusterBrokersOverviewVO
.
getKafkaRoleList
().
add
(
KafkaConstant
.
CONTROLLER_ROLE
);
}
clusterBrokersOverviewVO
.
setLatestMetrics
(
brokerMetrics
);
return
clusterBrokersOverviewVO
;
...
...
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/constant/KafkaConstant.java
浏览文件 @
949d6ba6
...
...
@@ -41,6 +41,8 @@ public class KafkaConstant {
public
static
final
Long
POLL_ONCE_TIMEOUT_UNIT_MS
=
2000L
;
public
static
final
String
CONTROLLER_ROLE
=
"controller"
;
public
static
final
Map
<
String
,
ConfigDef
.
ConfigKey
>
KAFKA_ALL_CONFIG_DEF_MAP
=
new
ConcurrentHashMap
<>();
static
{
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录