8x48/KnowStreaming
1
0
Fork 0
mirror of https://github.com/didi/KnowStreaming.git synced 2025-12-24 11:52:08 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge pull request #2 from didi/master

Browse Source 
sync code
This commit is contained in:
Hongten
2021-07-04 18:04:55 +08:00
committed by GitHub
parent 31fc6e4e56 1cd524c0cc
commit 48d2733403
76 changed files with 1045 additions and 365 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
README.md
View File

@@ -14,7 +14,8 @@
滴滴Logi-KafkaManager脱胎于滴滴内部多年的Kafka运营实践经验，是面向Kafka用户、Kafka运维人员打造的共享多租户Kafka云平台。专注于Kafka运维管控、监控告警、资源治理等核心场景，经历过大规模集群、海量大数据的考验。内部满意度高达90%的同时，还与多家知名企业达成商业化合作。
### 1.1 快速体验地址
- 体验地址 http://117.51.146.109:8080 账号密码 admin/admin
- 体验地址 http://117.51.150.133:8080 账号密码 admin/admin
### 1.2 体验地图
相比较于同类产品的用户视角单一（大多为管理员视角），滴滴Logi-KafkaManager建立了基于分角色、多场景视角的体验地图。分别是：**用户体验地图、运维体验地图、运营体验地图**

44
Releases_Notes.md
View File

@@ -7,6 +7,50 @@
---
## v2.4.1+
版本上线时间：2021-05-21
### 能力提升
- 增加直接增加权限和配额的接口(v2.4.1)
- 增加接口调用可绕过登录的功能(v2.4.1)
### 体验优化
- tomcat 版本提升至8.5.66(v2.4.2)
- op接口优化，拆分util接口为topic、leader两类接口(v2.4.1)
- 简化Gateway配置的Key长度(v2.4.1)
### bug修复
- 修复页面展示版本错误问题(v2.4.2)
## v2.4.0
版本上线时间：2021-05-18
### 能力提升
- 增加App与Topic自动化审批开关
- Broker元信息中增加Rack信息
- 升级MySQL 驱动，支持MySQL 8+
- 增加操作记录查询界面
### 体验优化
- FAQ告警组说明优化
- 用户手册共享及 独享集群概念优化
- 用户管理界面，前端限制用户删除自己
### bug修复
- 修复op-util类中创建Topic失败的接口
- 周期同步Topic到DB的任务修复，将Topic列表查询从缓存调整为直接查DB
- 应用下线审批失败的功能修复，将权限为0(无权限)的数据进行过滤
- 修复登录及权限绕过的漏洞
- 修复研发角色展示接入集群、暂停监控等按钮的问题
## v2.3.0
版本上线时间：2021-02-08

2
build.sh
View File

@@ -4,7 +4,7 @@ cd $workspace
## constant
OUTPUT_DIR=./output
KM_VERSION=2.4.0
KM_VERSION=2.4.2
APP_NAME=kafka-manager
APP_DIR=${APP_NAME}-${KM_VERSION}

48
docs/dev_guide/dynamic_config_manager.md
View File

@@ -14,6 +14,8 @@
- 1、Topic定时同步任务
- 2、专家服务——Topic分区热点
- 3、专家服务——Topic分区不足
- 4、专家服务——Topic资源治理
- 5、账单配置
## 1、Topic定时同步任务
@@ -119,4 +121,48 @@ TOPIC_INSUFFICIENT_PARTITION_CONFIG
50
]
}
```
```
## 4、专家服务——Topic资源治理
首先，我们认为在一定的时间长度内，Topic的分区offset没有任何变化的Topic，即没有数据写入的Topic，为过期的Topic。
Topic分区不足相关的动态配置(页面在运维管控->平台管理->配置管理)：
配置Key：
```
EXPIRED_TOPIC_CONFIG
```
配置Value：
```json
{
"minExpiredDay": 30, #过期时间大于此值才显示
"ignoreClusterIdList": [ # 忽略的集群
50
]
}
```
## 5、账单配置
Logi-KafkaManager除了作为Kafka运维管控平台之外，实际上还会有一些资源定价相关的功能。
当前定价方式：当月Topic的maxAvgDay天的峰值的均值流量作为Topic的使用额度。使用的额度 * 单价 * 溢价(预留buffer) 就等于当月的费用。
详细的计算逻辑见：com.xiaojukeji.kafka.manager.task.dispatch.biz.CalKafkaTopicBill; 和 com.xiaojukeji.kafka.manager.task.dispatch.biz.CalTopicStatistics;
这块在计算Topic的费用的配置如下所示：
配置Key：
```
KAFKA_TOPIC_BILL_CONFIG
```
配置Value：
```json
{
"maxAvgDay": 10, # 使用额度的计算规则
"quotaRatio": 1.5, # 溢价率
"priseUnitMB": 100 # 单价，即单MB/s流量多少钱
}
```

10
docs/install_guide/create_mysql_table.sql
View File

@@ -210,11 +210,11 @@ CREATE TABLE `gateway_config` (
PRIMARY KEY (`id`),
UNIQUE KEY `uniq_type_name` (`type`,`name`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COMMENT='gateway配置';
INSERT INTO gateway_config(type, name, value, `version`) values('SERVICE_DISCOVERY_QUEUE_SIZE', 'SERVICE_DISCOVERY_QUEUE_SIZE', 100000000, 1);
INSERT INTO gateway_config(type, name, value, `version`) values('SERVICE_DISCOVERY_APPID_RATE', 'SERVICE_DISCOVERY_APPID_RATE', 100000000, 1);
INSERT INTO gateway_config(type, name, value, `version`) values('SERVICE_DISCOVERY_IP_RATE', 'SERVICE_DISCOVERY_IP_RATE', 100000000, 1);
INSERT INTO gateway_config(type, name, value, `version`) values('SERVICE_DISCOVERY_SP_RATE', 'app_01234567', 100000000, 1);
INSERT INTO gateway_config(type, name, value, `version`) values('SERVICE_DISCOVERY_SP_RATE', '192.168.0.1', 100000000, 1);
INSERT INTO gateway_config(type, name, value, `version`, `description`) values('SD_QUEUE_SIZE', 'SD_QUEUE_SIZE', 100000000, 1, '任意集群队列大小');
INSERT INTO gateway_config(type, name, value, `version`, `description`) values('SD_APP_RATE', 'SD_APP_RATE', 100000000, 1, '任意一个App限速');
INSERT INTO gateway_config(type, name, value, `version`, `description`) values('SD_IP_RATE', 'SD_IP_RATE', 100000000, 1, '任意一个IP限速');
INSERT INTO gateway_config(type, name, value, `version`, `description`) values('SD_SP_RATE', 'app_01234567', 100000000, 1, '指定App限速');
INSERT INTO gateway_config(type, name, value, `version`, `description`) values('SD_SP_RATE', '192.168.0.1', 100000000, 1, '指定IP限速');
--
-- Table structure for table `heartbeat`

49
docs/user_guide/call_api_bypass_login.md Normal file
View File

@@ -0,0 +1,49 @@
---
![kafka-manager-logo](../assets/images/common/logo_name.png)
**一站式`Apache Kafka`集群指标监控与运维管控平台**
---
# 登录绕过
## 背景
现在除了开放出来的第三方接口，其他接口都需要走登录认证。
但是第三方接口不多，开放出来的能力有限，但是登录的接口又需要登录，非常的麻烦。
因此，新增了一个登录绕过的功能，为一些紧急临时的需求，提供一个调用不需要登录的能力。
## 使用方式
步骤一：接口调用时，在header中，增加如下信息：
```shell
# 表示开启登录绕过
Trick-Login-Switch : on
# 登录绕过的用户, 这里可以是admin, 或者是其他的, 但是必须在运维管控->平台管理->用户管理中设置了该用户。
Trick-Login-User : admin
```
 
步骤二：在运维管控->平台管理->平台配置上，设置允许了该用户以绕过的方式登录
```shell
# 设置的key，必须是这个
SECURITY.TRICK_USERS
# 设置的value，是json数组的格式，例如
[ "admin", "logi"]
```
 
步骤三：解释说明
设置完成上面两步之后，就可以直接调用需要登录的接口了。
但是还有一点需要注意，绕过的用户仅能调用他有权限的接口，比如一个普通用户，那么他就只能调用普通的接口，不能去调用运维人员的接口。

84
docs/user_guide/faq.md
View File

@@ -23,6 +23,14 @@
- 11、连接信息、耗时信息、磁盘信息为什么没有数据？
- 12、逻辑集群申请审批通过之后为什么看不到逻辑集群？
- 13、heartbeat表关联业务和使用场景是什么？
- 14、集群的删除，是否会真正的删除集群？
- 15、APP(应用)如何被使用起来？
- 16、为什么下线应用提示operation forbidden？
- 17、删除Topic成功，为什么过一会儿之后又出现了？
- 18、如何在不登录的情况下，调用一些需要登录的接口？
- 19、为什么无法看到连接信息、耗时信息等指标？
- 20、AppID鉴权、生产消费配额不起作用
---
### 0、支持哪些Kafka版本？
@@ -129,3 +137,79 @@
逻辑集群的申请与审批仅仅只是一个工单流程，并不会去实际创建逻辑集群，逻辑集群的创建还需要手动去创建。
具体的操作可见：[kafka-manager 接入集群](add_cluster/add_cluster.md)。
### 13、heartbeat表关联业务和使用场景是什么？
做任务抢占用的。
KM支持HA的方式部署，那么部署多台的时候，就会出现每一台都可能去做指标收集的事情，这块就使用heartbeat表做KM的存活性判断，然后进行任务的抢占或者是均衡。
更多详细的内容，可以看一下源码中，heartbeat表在哪里被使用了。
### 14、集群的删除，是否会真正的删除集群？
Logi-KM的运维管控，集群列表中的集群删除，仅仅只是将该集群从Logi-KM中进行删除，并不会对真正的物理集群做什么操作。
### 15、APP(应用)如何被使用起来？
app在Logi-KM中可以近似理解为租户，或者是kafka里面的一个账号的概念。
界面中显示的app信息、权限信息等，在平台层面仅仅只是控制Topic或集群在平台上的可见性，如果使用的是社区版本的Kafka，那么实际上是不能真正的管控到客户端对Topic的生产和消费。
但是如果是使用的滴滴的Kafka-Gateway，那么是可以做到对客户端的生产和消费的权限管控。滴滴的Kafka-Gateway暂未开源，属于企业服务，具体的可以入群交流，群地址在README中。
### 16、为什么下线应用提示operation forbidden？
**原因一：**
该应用还存在对Topic的权限，因此导致下线失败。具体查看的位置在"Topic管理-》应用管理-》详情"，可以看到应用对哪些Topic还有权限。
只有当权限全部去除之后，才可以下线应用。
**原因二：**
使用的是2.4.0之前的旧版本，旧版本存在缓存更新的BUG，建议升级至最新的版本，或者简单粗暴的就是重启一下KM。
### 17、删除Topic成功，为什么过一会儿之后又出现了？
**原因说明：**
Logi-KM会去请求Topic的endoffset信息，要获取这个信息就需要发送metadata请求，发送metadata请求的时候，如果集群允许自动创建Topic，那么当Topic不存在时，就会自动将该Topic创建出来。
**问题解决：**
因为在Logi-KM上，禁止Kafka客户端内部元信息获取这个动作非常的难做到，因此短时间内这个问题不好从Logi-KM上解决。
当然，对于不存在的Topic，Logi-KM是不会进行元信息请求的，因此也不用担心会莫名其妙的创建一个Topic出来。
但是，另外一点，对于开启允许Topic自动创建的集群，建议是关闭该功能，开启是非常危险的，如果关闭之后，Logi-KM也不会有这个问题。
最后这里举个开启这个配置后，非常危险的代码例子吧：
```java
for (int i= 0; i < 100000; ++i) {
// 如果是客户端类似这样写的，那么一启动，那么将创建10万个Topic出来，集群元信息瞬间爆炸，controller可能就不可服务了。
producer.send(new ProducerRecord<String, String>("logi_km" + i,"hello logi_km"));
}
```
### 18、如何在不登录的情况下，调用一些需要登录的接口？
具体见：[登录绕过](./call_api_bypass_login.md)
### 19、为什么无法看到连接信息、耗时信息等指标？
连接信息、耗时信息等指标依赖于滴滴kafka-gateway和滴滴Kafka引擎，通过gateway可获取到连接到该Topic的应用情况，提高对Topic的管控能力。通过滴滴Kafka引擎的自带埋点，可获取到耗时信息，提升Topic生产消费时的可观测性。这部分内容是属于商业版的范畴，暂未开源。如有需要，可进行商业合作。
具体见：[滴滴Logi-KafkaManager开源版和商业版特性对比](../开源版与商业版特性对比.md)
### 20、AppID鉴权、生产消费配额不起作用？
AppID鉴权、生产消费配额依赖于滴滴kafka-gateway，通过gateway进行身份鉴权和生产消费限流，可避免用户无限制的使用集群的流量，流量大的用户会耗尽系统资源从而影响其他用户的使用，造成集群的节点故障。这部分内容是属于商业版的范畴，暂未开源。如有需要，可进行商业合作。
具体见：[滴滴Logi-KafkaManager开源版和商业版特性对比](../开源版与商业版特性对比.md)

9
docs/user_guide/alarm_rules.md → docs/user_guide/monitor_desc.md
View File

@@ -4,11 +4,16 @@
---
## 报警策略-监控指标说明
| 指标 | 含义 |备注 |
| --- | --- | --- |
| online-kafka-consumer-lag | 消费时，按照分区的维度进行监控lag数 | lag表示有多少数据没有被消费，因为按照分区的维度监控，所以告警时一般会有分区信息 |
| online-kafka-consumer-maxLag | 消费时，按照整个Topic的维度，监控Topic所有的分区里面的那个最大的lag | 比如每个分区的lag分别是3、5、7，那么maxLag的值就是max(3,5,7)=7 |
| online-kafka-consumer-maxDelayTime | 消费时，按照Topic维度监控预计的消费延迟 | 这块是按照lag和messagesIn之间的关系计算出来的，可能会有误差 |
## 报警策略-报警函数介绍
| 类别 | 函数 | 含义 |函数文案 |备注 |
| --- | --- | --- | --- | --- |
| 发生次数 |all，n | 最近$n个周期内，全发生 | 连续发生(all) | |

55
docs/开源版与商业版特性对比.md Normal file
View File

@@ -0,0 +1,55 @@
---
![kafka-manager-logo](assets/images/common/logo_name.png)
**一站式`Apache Kafka`集群指标监控与运维管控平台**
---
**开源版、商业版对比**
纲要：Logi-KafakManager的商业特性是强依赖于滴滴Kafka Gateway和滴滴kafka引擎。
滴滴KafkaGateway主要负责：服务发现、安全管控（身份鉴权、生产消费鉴权等）、流量管控（应用配额等）等；
滴滴Kafka引擎主要负责：更丰富的监控指标（broker实时耗时、压缩指标、分区落盘等）、磁盘过载保护等
备注：两个版本的产品功能页面是一样的。区别在于开源版未使用滴滴KafkaGateway（滴滴Kafka引擎），部分产品功能/功能不起作用或者页面无数据
| 模块 |对比指标 |底层依赖 |开源版 |商业版 |备注 |
| --- | --- | --- | --- | --- | --- |
| 服务发现 | bootstrap地址变更对客户端无影响 | | | 是| |
| 安全管控 | 身份鉴权（appID+password） | | | 是 | |
| | 权限鉴权（Topic+appID） | | | 是 | |
| 指标监控 | Topic实时流量、历史流量 | | 是 | 是 | |
| | Broker实时耗时、历史耗时 | 引擎 | | 是 | |
| | 分区落盘 | 引擎 | | 是 | |
| | Topic里的数据压缩格式 | 引擎 | | 是 | |
| | 连接信息（Topic上有哪些应用连接） | Gateway | | 是| |
| | 流量管控（应用配额、生产消费限流等） | Gateway | | 是 | |
| 监控报警 | | | 是 | 是 | 监控指标上报，需对接外部监控系统（夜莺or企业内部监控系统） |
| Topic运维 | 申请分区 | | 是 | 是 | |
| | 调整配额 | Gateway | | | 是 |
| | Topic数据采样 | | 是 | 是 | |
| | 消费组管理（重置消费偏移等） | | 是 | 是 | |
| 集群管理 | 集群接入（部署） | | 是 | 是 | 需手动部署集群，或借助外部的自动化部署系统（夜莺）来部署系统 |
| | 集群指标监控 | | 是 | 是 | |
| | 按照Region、逻辑集群进行管理 | | 是 | 是 | |
| | Topic迁移 | | 是 | 是| |
| | 集群任务（集群版本管理、升级、扩缩容、回滚等） | | 是 | 是 | 需借助夜莺或自动化部署系统来实现 |
| | 磁盘过载保护 | 引擎 | | 是 | |
| | 指定broker作为优选controller | Gateway | | 是 | |
| Gateway管理 | 管理 Gateway的配置文件 | Gateway | | 是 | |
| 资源治理 | 专家服务（Topic分区热点、Topic分区不足、Topic长期未使用、Topic流量异常） | | 是 | 是 | 开源版：具备问题发现与基础的问题解决能力；商业版：可在此基础上，融入滴滴内部的资源治理经验，提供更加专家化的问题解决方法 |
| | 健康分 | | 是 | 是 | 开源版：具备基础的健康分算法；商业版：可融入更多的指标统计，及定制化的健康分算法 |
| 运营管理 | 资源审批（应用申请、Topic申请、分区申请、配额申请、集群申请等，都需要通过工单进行审批） | |是 | 是 | |
| | 账单体系（根据流量核算Topic、集群费用） | | 是 | 是| |
**总结**
Logi-KafkaManager的商业特性体现在在滴滴Kafka Gateway、滴滴Kafka引擎、内部沉淀出的资源治理专家经验、可定制化的健康分算法。
从场景来看，滴滴Logi-KafkaManager的开源版本在kafka集群运维、的Topic管理、监控告警、资源治理等kafka核心场景都充分开源用户的使用需求并且有着出色的表现。而商业版相较于开源版在安全管控、流量管控、更丰富的指标监控、资源治理专家经验的具有明显提升，更加符合企业业务需求。
除此之外，商业版还可根据企业实际需求对平台源码进行定制化改造，并提供运维保障，稳定性保障，运营保障等服务。

19
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/bizenum/ApiLevelEnum.java
View File

@@ -1,19 +0,0 @@
package com.xiaojukeji.kafka.manager.common.bizenum;
/**
* @author zengqiao
* @date 20/7/27
*/
public enum ApiLevelEnum {
LEVEL_0(0),
LEVEL_1(1),
LEVEL_2(2),
LEVEL_3(3)
;
private int level;
ApiLevelEnum(int level) {
this.level = level;
}
}

8
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/bizenum/ConsumeHealthEnum.java
View File

@@ -24,18 +24,10 @@ public enum ConsumeHealthEnum {
return code;
}
public void setCode(Integer code) {
this.code = code;
}
public String getMessage() {
return message;
}
public void setMessage(String message) {
this.message = message;
}
@Override
public String toString() {
return "ConsumeHealthEnum{" +

7
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/bizenum/DBStatusEnum.java
View File

@@ -19,7 +19,10 @@ public enum DBStatusEnum {
return status;
}
public void setStatus(int status) {
this.status = status;
@Override
public String toString() {
return "DBStatusEnum{" +
"status=" + status +
'}';
}
}

8
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/bizenum/IDCEnum.java
View File

@@ -21,18 +21,10 @@ public enum IDCEnum {
return idc;
}
public void setIdc(String idc) {
this.idc = idc;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
@Override
public String toString() {
return "IDCEnum{" +

4
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/bizenum/KafkaBrokerRoleEnum.java
View File

@@ -21,10 +21,6 @@ public enum KafkaBrokerRoleEnum {
return role;
}
public void setRole(String role) {
this.role = role;
}
@Override
public String toString() {
return "KafkaBrokerRoleEnum{" +

8
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/bizenum/KafkaClientEnum.java
View File

@@ -24,18 +24,10 @@ public enum KafkaClientEnum {
return code;
}
public void setCode(Integer code) {
this.code = code;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
@Override
public String toString() {
return "KafkaClientEnum{" +

7
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/bizenum/OffsetResetTypeEnum.java
View File

@@ -18,4 +18,11 @@ public enum OffsetResetTypeEnum {
public Integer getCode() {
return code;
}
@Override
public String toString() {
return "OffsetResetTypeEnum{" +
"code=" + code +
'}';
}
}

8
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/bizenum/OperationStatusEnum.java
View File

@@ -27,4 +27,12 @@ public enum OperationStatusEnum {
public String getMessage() {
return message;
}
@Override
public String toString() {
return "OperationStatusEnum{" +
"code=" + code +
", message='" + message + '\'' +
'}';
}
}

12
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/bizenum/PeakFlowStatusEnum.java
View File

@@ -15,9 +15,9 @@ public enum PeakFlowStatusEnum {
;
public Integer code;
private Integer code;
public String message;
private String message;
PeakFlowStatusEnum(Integer code, String message) {
this.code = code;
@@ -28,18 +28,10 @@ public enum PeakFlowStatusEnum {
return code;
}
public void setCode(Integer code) {
this.code = code;
}
public String getMessage() {
return message;
}
public void setMessage(String message) {
this.message = message;
}
@Override
public String toString() {
return "PeakFlowStatusEnum{" +

8
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/bizenum/RebalanceDimensionEnum.java
View File

@@ -29,4 +29,12 @@ public enum RebalanceDimensionEnum {
public String getMessage() {
return message;
}
@Override
public String toString() {
return "RebalanceDimensionEnum{" +
"code=" + code +
", message='" + message + '\'' +
'}';
}
}

13
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/bizenum/TaskStatusEnum.java
View File

@@ -43,18 +43,10 @@ public enum TaskStatusEnum {
return code;
}
public void setCode(Integer code) {
this.code = code;
}
public String getMessage() {
return message;
}
public void setMessage(String message) {
this.message = message;
}
@Override
public String toString() {
return "TaskStatusEnum{" +
@@ -64,9 +56,6 @@ public enum TaskStatusEnum {
}
public static Boolean isFinished(Integer code) {
if (code >= FINISHED.getCode()) {
return true;
}
return false;
return code >= FINISHED.getCode();
}
}

7
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/bizenum/TaskStatusReassignEnum.java
View File

@@ -45,11 +45,6 @@ public enum TaskStatusReassignEnum {
}
public static Boolean isFinished(Integer code) {
if (SUCCEED.getCode().equals(code)
|| FAILED.getCode().equals(code)
|| CANCELED.getCode().equals(code)) {
return true;
}
return false;
return SUCCEED.getCode().equals(code) || FAILED.getCode().equals(code) || CANCELED.getCode().equals(code);
}
}

8
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/bizenum/TopicAuthorityEnum.java
View File

@@ -33,4 +33,12 @@ public enum TopicAuthorityEnum {
public String getMessage() {
return message;
}
@Override
public String toString() {
return "TopicAuthorityEnum{" +
"code=" + code +
", message='" + message + '\'' +
'}';
}
}

8
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/bizenum/TopicExpiredStatusEnum.java
View File

@@ -29,4 +29,12 @@ public enum TopicExpiredStatusEnum {
public String getMessage() {
return message;
}
@Override
public String toString() {
return "TopicExpiredStatusEnum{" +
"status=" + status +
", message='" + message + '\'' +
'}';
}
}

8
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/bizenum/TopicOffsetChangedEnum.java
View File

@@ -23,18 +23,10 @@ public enum TopicOffsetChangedEnum {
return code;
}
public void setCode(Integer code) {
this.code = code;
}
public String getMessage() {
return message;
}
public void setMessage(String message) {
this.message = message;
}
@Override
public String toString() {
return "TopicOffsetChangedEnum{" +

18
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/bizenum/gateway/GatewayConfigKeyEnum.java
View File

@@ -5,11 +5,11 @@ package com.xiaojukeji.kafka.manager.common.bizenum.gateway;
* @date 20/7/28
*/
public enum GatewayConfigKeyEnum {
SD_CLUSTER_ID("SERVICE_DISCOVERY_CLUSTER_ID", "SERVICE_DISCOVERY_CLUSTER_ID"),
SD_QUEUE_SIZE("SERVICE_DISCOVERY_QUEUE_SIZE", "SERVICE_DISCOVERY_QUEUE_SIZE"),
SD_APP_ID_RATE("SERVICE_DISCOVERY_APPID_RATE", "SERVICE_DISCOVERY_APPID_RATE"),
SD_IP_RATE("SERVICE_DISCOVERY_IP_RATE", "SERVICE_DISCOVERY_IP_RATE"),
SD_SP_RATE("SERVICE_DISCOVERY_SP_RATE", "SERVICE_DISCOVERY_SP_RATE"),
SD_CLUSTER_ID("SD_CLUSTER_ID", "SD_CLUSTER_ID"),
SD_QUEUE_SIZE("SD_QUEUE_SIZE", "SD_QUEUE_SIZE"),
SD_APP_RATE("SD_APP_RATE", "SD_APP_RATE"),
SD_IP_RATE("SD_IP_RATE", "SD_IP_RATE"),
SD_SP_RATE("SD_SP_RATE", "SD_SP_RATE"),
;
@@ -26,18 +26,10 @@ public enum GatewayConfigKeyEnum {
return configType;
}
public void setConfigType(String configType) {
this.configType = configType;
}
public String getConfigName() {
return configName;
}
public void setConfigName(String configName) {
this.configName = configName;
}
@Override
public String toString() {
return "GatewayConfigKeyEnum{" +

3
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/constant/ApiLevelContent.java
View File

@@ -12,4 +12,7 @@ public class ApiLevelContent {
public static final int LEVEL_NORMAL_3 = 3;
public static final int LEVEL_DEFAULT_4 = 4;
private ApiLevelContent() {
}
}

3
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/constant/ApiPrefix.java
View File

@@ -28,4 +28,7 @@ public class ApiPrefix {
// gateway
public static final String GATEWAY_API_V1_PREFIX = "/gateway" + API_V1_PREFIX;
private ApiPrefix() {
}
}

3
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/constant/ConfigConstant.java
View File

@@ -30,4 +30,7 @@ public class ConfigConstant {
public static final String BROKER_CAPACITY_LIMIT_CONFIG_KEY = "BROKER_CAPACITY_LIMIT_CONFIG";
public static final String KAFKA_CLUSTER_DO_CONFIG_KEY = "KAFKA_CLUSTER_DO_CONFIG";
private ConfigConstant() {
}
}

3
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/constant/Constant.java
View File

@@ -45,4 +45,7 @@ public class Constant {
public static final Integer DEFAULT_MAX_CAL_TOPIC_EXPIRED_DAY = 90;
public static final Integer INVALID_CODE = -1;
private Constant() {
}
}

3
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/constant/KafkaConstant.java
View File

@@ -16,4 +16,7 @@ public class KafkaConstant {
public static final String CLIENT_VERSION_NAME_UNKNOWN = "unknown";
public static final String RETENTION_MS_KEY = "retention.ms";
private KafkaConstant() {
}
}

3
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/constant/KafkaMetricsCollections.java
View File

@@ -39,4 +39,7 @@ public class KafkaMetricsCollections {
* Broker信息
*/
public static final int BROKER_VERSION = 400;
private KafkaMetricsCollections() {
}
}

3
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/constant/LogConstant.java
View File

@@ -10,4 +10,7 @@ public class LogConstant {
public static final String API_METRICS_LOGGER = "API_METRICS_LOGGER";
public static final String SCHEDULED_TASK_LOGGER = "SCHEDULED_TASK_LOGGER";
private LogConstant() {
}
}

3
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/constant/LoginConstant.java
View File

@@ -11,4 +11,7 @@ public class LoginConstant {
public static final String COOKIE_CHINESE_USERNAME_KEY = "chineseName";
public static final Integer COOKIE_OR_SESSION_MAX_AGE_UNIT_MS = 24 * 60 * 60 * 1000;
private LoginConstant() {
}
}

3
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/constant/SystemCodeConstant.java
View File

@@ -6,4 +6,7 @@ package com.xiaojukeji.kafka.manager.common.constant;
*/
public class SystemCodeConstant {
public static final String KAFKA_MANAGER = "kafka-manager";
private SystemCodeConstant() {
}
}

5
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/constant/TopicCreationConstant.java
View File

@@ -25,6 +25,8 @@ public class TopicCreationConstant {
public static final String TOPIC_RETENTION_TIME_KEY_NAME = "retention.ms";
public static final String TOPIC_RETENTION_BYTES_KEY_NAME = "retention.bytes";
public static final Long DEFAULT_QUOTA = 3 * 1024 * 1024L;
public static Properties createNewProperties(Long retentionTime) {
@@ -54,4 +56,7 @@ public class TopicCreationConstant {
* 单次自动化审批, 最多允许的通过单子
*/
public static final Integer MAX_PASSED_ORDER_NUM_PER_TASK = 200;
private TopicCreationConstant() {
}
}

3
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/constant/TopicSampleConstant.java
View File

@@ -16,4 +16,7 @@ public class TopicSampleConstant {
public static final Integer MAX_TIMEOUT_UNIT_MS = 10000;
public static final Integer POLL_TIME_OUT_UNIT_MS = 2000;
public static final Integer MAX_DATA_LENGTH_UNIT_BYTE = 2048;
private TopicSampleConstant() {
}
}

24
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/constant/TrickLoginConstant.java Normal file
View File

@@ -0,0 +1,24 @@
package com.xiaojukeji.kafka.manager.common.constant;
public class TrickLoginConstant {
/**
* HTTP Header key
*/
public static final String TRICK_LOGIN_SWITCH = "Trick-Login-Switch";
public static final String TRICK_LOGIN_USER = "Trick-Login-User";
/**
* 配置允许 trick 登录用户名单
*/
public static final String TRICK_LOGIN_LEGAL_USER_CONFIG_KEY = "SECURITY.TRICK_USERS";
/**
* 开关状态值
*/
public static final String TRICK_LOGIN_SWITCH_ON = "on";
public static final String TRICK_LOGIN_SWITCH_OFF = "off";
private TrickLoginConstant() {
}
}

13
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/entity/ao/gateway/TopicQuota.java
View File

@@ -1,5 +1,7 @@
package com.xiaojukeji.kafka.manager.common.entity.ao.gateway;
import com.xiaojukeji.kafka.manager.common.entity.dto.gateway.TopicQuotaDTO;
/**
* @author zhongyuankai
* @date 2020/4/27
@@ -65,4 +67,15 @@ public class TopicQuota {
", consumeQuota=" + consumeQuota +
'}';
}
public static TopicQuota buildFrom(TopicQuotaDTO dto) {
TopicQuota topicQuota = new TopicQuota();
topicQuota.setAppId(dto.getAppId());
topicQuota.setClusterId(dto.getClusterId());
topicQuota.setTopicName(dto.getTopicName());
topicQuota.setProduceQuota(dto.getProduceQuota());
topicQuota.setConsumeQuota(dto.getConsumeQuota());
return topicQuota;
}
}

13
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/entity/ao/topic/TopicBasicDTO.java
View File

@@ -37,6 +37,8 @@ public class TopicBasicDTO {
private Long retentionTime;
private Long retentionBytes;
public Long getClusterId() {
return clusterId;
}
@@ -157,6 +159,14 @@ public class TopicBasicDTO {
this.retentionTime = retentionTime;
}
public Long getRetentionBytes() {
return retentionBytes;
}
public void setRetentionBytes(Long retentionBytes) {
this.retentionBytes = retentionBytes;
}
@Override
public String toString() {
return "TopicBasicDTO{" +
@@ -166,7 +176,7 @@ public class TopicBasicDTO {
", principals='" + principals + '\'' +
", topicName='" + topicName + '\'' +
", description='" + description + '\'' +
", regionNameList='" + regionNameList + '\'' +
", regionNameList=" + regionNameList +
", score=" + score +
", topicCodeC='" + topicCodeC + '\'' +
", partitionNum=" + partitionNum +
@@ -175,6 +185,7 @@ public class TopicBasicDTO {
", modifyTime=" + modifyTime +
", createTime=" + createTime +
", retentionTime=" + retentionTime +
", retentionBytes=" + retentionBytes +
'}';
}
}

47
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/entity/dto/gateway/TopicQuotaDTO.java Normal file
View File

@@ -0,0 +1,47 @@
package com.xiaojukeji.kafka.manager.common.entity.dto.gateway;
import com.xiaojukeji.kafka.manager.common.entity.dto.ClusterTopicDTO;
import com.xiaojukeji.kafka.manager.common.utils.ValidateUtils;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
@ApiModel(description = "配额调整")
public class TopicQuotaDTO extends ClusterTopicDTO {
@ApiModelProperty(value = "appId")
private String appId;
@ApiModelProperty(value = "发送数据速率B/s")
private Long produceQuota;
@ApiModelProperty(value = "消费数据速率B/s")
private Long consumeQuota;
public String getAppId() {
return appId;
}
public void setAppId(String appId) {
this.appId = appId;
}
public Long getProduceQuota() {
return produceQuota;
}
public void setProduceQuota(Long produceQuota) {
this.produceQuota = produceQuota;
}
public Long getConsumeQuota() {
return consumeQuota;
}
public void setConsumeQuota(Long consumeQuota) {
this.consumeQuota = consumeQuota;
}
@Override
public boolean paramLegal() {
return !ValidateUtils.isNullOrLessThanZero(clusterId) && !ValidateUtils.isBlank(topicName) && !ValidateUtils.isBlank(appId);
}
}

108
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/entity/vo/normal/topic/TopicBasicVO.java
View File

@@ -33,6 +33,9 @@ public class TopicBasicVO {
@ApiModelProperty(value = "存储时间(ms)")
private Long retentionTime;
@ApiModelProperty(value = "单分区数据保存大小(Byte)")
private Long retentionBytes;
@ApiModelProperty(value = "创建时间")
private Long createTime;
@@ -62,12 +65,20 @@ public class TopicBasicVO {
this.clusterId = clusterId;
}
public String getTopicCodeC() {
return topicCodeC;
public String getAppId() {
return appId;
}
public void setTopicCodeC(String topicCodeC) {
this.topicCodeC = topicCodeC;
public void setAppId(String appId) {
this.appId = appId;
}
public String getAppName() {
return appName;
}
public void setAppName(String appName) {
this.appName = appName;
}
public Integer getPartitionNum() {
@@ -86,22 +97,6 @@ public class TopicBasicVO {
this.replicaNum = replicaNum;
}
public Long getModifyTime() {
return modifyTime;
}
public void setModifyTime(Long modifyTime) {
this.modifyTime = modifyTime;
}
public Long getCreateTime() {
return createTime;
}
public void setCreateTime(Long createTime) {
this.createTime = createTime;
}
public String getPrincipals() {
return principals;
}
@@ -110,30 +105,6 @@ public class TopicBasicVO {
this.principals = principals;
}
public String getDescription() {
return description;
}
public void setDescription(String description) {
this.description = description;
}
public void setAppId(String appId) {
this.appId = appId;
}
public void setBootstrapServers(String bootstrapServers) {
this.bootstrapServers = bootstrapServers;
}
public String getAppId() {
return appId;
}
public String getBootstrapServers() {
return bootstrapServers;
}
public Long getRetentionTime() {
return retentionTime;
}
@@ -142,12 +113,28 @@ public class TopicBasicVO {
this.retentionTime = retentionTime;
}
public String getAppName() {
return appName;
public Long getRetentionBytes() {
return retentionBytes;
}
public void setAppName(String appName) {
this.appName = appName;
public void setRetentionBytes(Long retentionBytes) {
this.retentionBytes = retentionBytes;
}
public Long getCreateTime() {
return createTime;
}
public void setCreateTime(Long createTime) {
this.createTime = createTime;
}
public Long getModifyTime() {
return modifyTime;
}
public void setModifyTime(Long modifyTime) {
this.modifyTime = modifyTime;
}
public Integer getScore() {
@@ -158,6 +145,30 @@ public class TopicBasicVO {
this.score = score;
}
public String getTopicCodeC() {
return topicCodeC;
}
public void setTopicCodeC(String topicCodeC) {
this.topicCodeC = topicCodeC;
}
public String getDescription() {
return description;
}
public void setDescription(String description) {
this.description = description;
}
public String getBootstrapServers() {
return bootstrapServers;
}
public void setBootstrapServers(String bootstrapServers) {
this.bootstrapServers = bootstrapServers;
}
public List<String> getRegionNameList() {
return regionNameList;
}
@@ -176,6 +187,7 @@ public class TopicBasicVO {
", replicaNum=" + replicaNum +
", principals='" + principals + '\'' +
", retentionTime=" + retentionTime +
", retentionBytes=" + retentionBytes +
", createTime=" + createTime +
", modifyTime=" + modifyTime +
", score=" + score +

24
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/utils/SpringTool.java
View File

@@ -2,6 +2,7 @@ package com.xiaojukeji.kafka.manager.common.utils;
import com.xiaojukeji.kafka.manager.common.constant.Constant;
import com.xiaojukeji.kafka.manager.common.constant.LoginConstant;
import com.xiaojukeji.kafka.manager.common.constant.TrickLoginConstant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
@@ -53,13 +54,6 @@ public class SpringTool implements ApplicationContextAware, DisposableBean {
return getApplicationContext().getBeansOfType(type);
}
// /**
// * 从静态变量applicationContext中去的Bean，自动转型为所复制对象的类型
// */
// public static <T> T getBean(Class<T> requiredType) {
// return (T) applicationContext.getBean(requiredType);
// }
/**
* 清除SpringContextHolder中的ApplicationContext为Null
*/
@@ -87,10 +81,18 @@ public class SpringTool implements ApplicationContextAware, DisposableBean {
}
public static String getUserName(){
HttpServletRequest request =
((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getRequest();
HttpSession session = request.getSession();
String username = (String) session.getAttribute(LoginConstant.SESSION_USERNAME_KEY);
HttpServletRequest request = ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getRequest();
String username = null;
if (TrickLoginConstant.TRICK_LOGIN_SWITCH_ON.equals(request.getHeader(TrickLoginConstant.TRICK_LOGIN_SWITCH))) {
// trick登录方式的获取用户
username = request.getHeader(TrickLoginConstant.TRICK_LOGIN_USER);
} else {
// 走页面登录方式登录的获取用户
HttpSession session = request.getSession();
username = (String) session.getAttribute(LoginConstant.SESSION_USERNAME_KEY);
}
if (ValidateUtils.isNull(username)) {
return Constant.DEFAULT_USER_NAME;
}

3
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/utils/jmx/JmxConnectorWrap.java
View File

@@ -79,7 +79,8 @@ public class JmxConnectorWrap {
try {
Map<String, Object> environment = new HashMap<String, Object>();
if (!ValidateUtils.isBlank(this.jmxConfig.getUsername()) && !ValidateUtils.isBlank(this.jmxConfig.getPassword())) {
environment.put(JMXConnector.CREDENTIALS, Arrays.asList(this.jmxConfig.getUsername(), this.jmxConfig.getPassword()));
// fixed by riyuetianmu
environment.put(JMXConnector.CREDENTIALS, new String[]{this.jmxConfig.getUsername(), this.jmxConfig.getPassword()});
}
if (jmxConfig.isOpenSSL() != null && this.jmxConfig.isOpenSSL()) {
environment.put(Context.SECURITY_PROTOCOL, "ssl");

3
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/zookeeper/ZkPathUtil.java
View File

@@ -119,4 +119,7 @@ public class ZkPathUtil {
public static String getControllerCandidatePath(Integer brokerId) {
return D_CONTROLLER_CANDIDATES + ZOOKEEPER_SEPARATOR + brokerId;
}
private ZkPathUtil() {
}
}

9
kafka-manager-console/package.json
View File

@@ -1,18 +1,18 @@
{
"name": "logi-kafka",
"version": "2.3.1",
"version": "2.4.3",
"description": "",
"scripts": {
"start": "webpack-dev-server",
"daily-build": "cross-env NODE_ENV=production webpack",
"pre-build": "cross-env NODE_ENV=production webpack",
"prod-build": "cross-env NODE_ENV=production webpack"
"prod-build": "cross-env NODE_ENV=production webpack",
"fix-memory": "cross-env LIMIT=4096 increase-memory-limit"
},
"author": "",
"license": "ISC",
"devDependencies": {
"@hot-loader/react-dom": "^16.8.6",
"@types/clipboard": "^2.0.1",
"@types/echarts": "^4.4.1",
"@types/lodash.debounce": "^4.0.6",
"@types/react": "^16.8.8",
@@ -21,12 +21,13 @@
"@types/spark-md5": "^3.0.2",
"antd": "^3.26.15",
"clean-webpack-plugin": "^3.0.0",
"clipboard": "2.0.6",
"clipboard": "^2.0.8",
"cross-env": "^7.0.2",
"css-loader": "^2.1.0",
"echarts": "^4.5.0",
"file-loader": "^5.0.2",
"html-webpack-plugin": "^3.2.0",
"increase-memory-limit": "^1.0.7",
"less": "^3.9.0",
"less-loader": "^4.1.0",
"mini-css-extract-plugin": "^0.6.0",

2
kafka-manager-console/src/container/alarm/add-alarm/alarm-select.tsx
View File

@@ -54,7 +54,7 @@ export class AlarmSelect extends React.Component<IAlarmSelectProps> {
<a
className="icon-color"
target="_blank"
href="https://github.com/didi/kafka-manager"
href="https://github.com/didi/Logi-KafkaManager/blob/master/docs/user_guide/faq.md"
>
新增规则组？
</a>

2
kafka-manager-console/src/container/header/index.tsx
View File

@@ -145,7 +145,7 @@ export const Header = observer((props: IHeader) => {
<div className="left-content">
<img className="kafka-header-icon" src={logoUrl} alt="" />
<span className="kafka-header-text">Kafka Manager</span>
<a className='kafka-header-version' href="https://github.com/didi/Logi-KafkaManager/releases" target='_blank'>v2.4.0</a>
<a className='kafka-header-version' href="https://github.com/didi/Logi-KafkaManager/releases" target='_blank'>v2.4.2</a>
{/* 添加版本超链接 */}
</div>
<div className="mid-content">

43
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/cache/PhysicalClusterMetadataManager.java vendored
View File

@@ -3,10 +3,12 @@ package com.xiaojukeji.kafka.manager.service.cache;
import com.xiaojukeji.kafka.manager.common.bizenum.KafkaBrokerRoleEnum;
import com.xiaojukeji.kafka.manager.common.constant.Constant;
import com.xiaojukeji.kafka.manager.common.constant.KafkaConstant;
import com.xiaojukeji.kafka.manager.common.constant.TopicCreationConstant;
import com.xiaojukeji.kafka.manager.common.entity.KafkaVersion;
import com.xiaojukeji.kafka.manager.common.entity.pojo.ClusterDO;
import com.xiaojukeji.kafka.manager.common.utils.JsonUtils;
import com.xiaojukeji.kafka.manager.common.utils.ListUtils;
import com.xiaojukeji.kafka.manager.common.utils.NumberUtils;
import com.xiaojukeji.kafka.manager.common.utils.ValidateUtils;
import com.xiaojukeji.kafka.manager.common.utils.jmx.JmxConfig;
import com.xiaojukeji.kafka.manager.common.utils.jmx.JmxConnectorWrap;
@@ -56,7 +58,7 @@ public class PhysicalClusterMetadataManager {
private final static Map<Long, Map<String, TopicMetadata>> TOPIC_METADATA_MAP = new ConcurrentHashMap<>();
private final static Map<Long, Map<String, Long>> TOPIC_RETENTION_TIME_MAP = new ConcurrentHashMap<>();
private final static Map<Long, Map<String, Properties>> TOPIC_PROPERTIES_MAP = new ConcurrentHashMap<>();
private final static Map<Long, Map<Integer, BrokerMetadata>> BROKER_METADATA_MAP = new ConcurrentHashMap<>();
@@ -95,7 +97,7 @@ public class PhysicalClusterMetadataManager {
// 初始化topic-map
TOPIC_METADATA_MAP.put(clusterDO.getId(), new ConcurrentHashMap<>());
TOPIC_RETENTION_TIME_MAP.put(clusterDO.getId(), new ConcurrentHashMap<>());
TOPIC_PROPERTIES_MAP.put(clusterDO.getId(), new ConcurrentHashMap<>());
// 初始化cluster-map
CLUSTER_MAP.put(clusterDO.getId(), clusterDO);
@@ -158,7 +160,7 @@ public class PhysicalClusterMetadataManager {
KAFKA_VERSION_MAP.remove(clusterId);
TOPIC_METADATA_MAP.remove(clusterId);
TOPIC_RETENTION_TIME_MAP.remove(clusterId);
TOPIC_PROPERTIES_MAP.remove(clusterId);
CLUSTER_MAP.remove(clusterId);
}
@@ -262,24 +264,45 @@ public class PhysicalClusterMetadataManager {
//---------------------------配置相关元信息--------------
public static void putTopicRetentionTime(Long clusterId, String topicName, Long retentionTime) {
Map<String, Long> timeMap = TOPIC_RETENTION_TIME_MAP.get(clusterId);
if (timeMap == null) {
public static void putTopicProperties(Long clusterId, String topicName, Properties properties) {
if (ValidateUtils.isNull(clusterId) || ValidateUtils.isBlank(topicName) || ValidateUtils.isNull(properties)) {
return;
}
timeMap.put(topicName, retentionTime);
Map<String, Properties> propertiesMap = TOPIC_PROPERTIES_MAP.get(clusterId);
if (ValidateUtils.isNull(propertiesMap)) {
return;
}
propertiesMap.put(topicName, properties);
}
public static Long getTopicRetentionTime(Long clusterId, String topicName) {
Map<String, Long> timeMap = TOPIC_RETENTION_TIME_MAP.get(clusterId);
if (timeMap == null) {
Map<String, Properties> propertiesMap = TOPIC_PROPERTIES_MAP.get(clusterId);
if (ValidateUtils.isNull(propertiesMap)) {
return null;
}
return timeMap.get(topicName);
Properties properties = propertiesMap.get(topicName);
if (ValidateUtils.isNull(properties)) {
return null;
}
return NumberUtils.string2Long(properties.getProperty(TopicCreationConstant.TOPIC_RETENTION_TIME_KEY_NAME));
}
public static Long getTopicRetentionBytes(Long clusterId, String topicName) {
Map<String, Properties> propertiesMap = TOPIC_PROPERTIES_MAP.get(clusterId);
if (ValidateUtils.isNull(propertiesMap)) {
return null;
}
Properties properties = propertiesMap.get(topicName);
if (ValidateUtils.isNull(properties)) {
return null;
}
return NumberUtils.string2Long(properties.getProperty(TopicCreationConstant.TOPIC_RETENTION_BYTES_KEY_NAME));
}
//---------------------------Broker元信息相关--------------

8
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/service/TopicManagerService.java
View File

@@ -10,6 +10,7 @@ import com.xiaojukeji.kafka.manager.common.entity.ao.topic.MineTopicSummary;
import com.xiaojukeji.kafka.manager.common.entity.pojo.TopicDO;
import com.xiaojukeji.kafka.manager.common.entity.pojo.TopicExpiredDO;
import com.xiaojukeji.kafka.manager.common.entity.pojo.TopicStatisticsDO;
import com.xiaojukeji.kafka.manager.common.entity.pojo.gateway.AuthorityDO;
import java.util.Date;
import java.util.List;
@@ -122,5 +123,12 @@ public interface TopicManagerService {
List<TopicStatisticsDO> getTopicStatistic(Long clusterId, String topicName, Date startTime, Date endTime);
TopicBusinessInfo getTopicBusinessInfo(Long physicalClusterId, String topicName);
/**
* topic权限调整
* @param authorityDO topic权限
* @return
*/
ResultStatus addAuthority(AuthorityDO authorityDO);
}

1
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/service/TopicService.java
View File

@@ -105,4 +105,5 @@ public interface TopicService {
List<TopicBrokerDTO> getTopicBrokerList(Long clusterId, String topicName);
Result<TopicOffsetChangedEnum> checkTopicOffsetChanged(Long physicalClusterId, String topicName, Long latestTime);
}

8
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/service/gateway/QuotaService.java
View File

@@ -1,5 +1,6 @@
package com.xiaojukeji.kafka.manager.service.service.gateway;
import com.xiaojukeji.kafka.manager.common.entity.ResultStatus;
import com.xiaojukeji.kafka.manager.common.entity.ao.gateway.TopicQuota;
/**
@@ -34,4 +35,11 @@ public interface QuotaService {
TopicQuota getQuotaFromZk(Long clusterId, String topicName, String appId);
Boolean modifyProduceQuota(Long clusterId, String topicName, String appId, Long produceQuota);
/**
* topic配额调整
* @param topicQuota topic配额
* @return
*/
ResultStatus addTopicQuotaByAuthority(TopicQuota topicQuota);
}

4
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/service/gateway/impl/GatewayConfigServiceImpl.java
View File

@@ -86,8 +86,8 @@ public class GatewayConfigServiceImpl implements GatewayConfigService {
GatewayConfigDO configDO = null;
try {
configDO = gatewayConfigDao.getByConfigTypeAndName(
GatewayConfigKeyEnum.SD_APP_ID_RATE.getConfigType(),
GatewayConfigKeyEnum.SD_APP_ID_RATE.getConfigName()
GatewayConfigKeyEnum.SD_APP_RATE.getConfigType(),
GatewayConfigKeyEnum.SD_APP_RATE.getConfigName()
);
if (ValidateUtils.isNull(configDO) || configDO.getVersion() <= requestVersion) {
return new AppRateConfig(Long.MIN_VALUE, null);

41
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/service/gateway/impl/QuotaServiceImpl.java
View File

@@ -1,11 +1,16 @@
package com.xiaojukeji.kafka.manager.service.service.gateway.impl;
import com.xiaojukeji.kafka.manager.common.bizenum.TopicAuthorityEnum;
import com.xiaojukeji.kafka.manager.common.entity.ResultStatus;
import com.xiaojukeji.kafka.manager.common.entity.ao.gateway.TopicQuota;
import com.xiaojukeji.kafka.manager.common.entity.pojo.gateway.AuthorityDO;
import com.xiaojukeji.kafka.manager.common.utils.NumberUtils;
import com.xiaojukeji.kafka.manager.common.utils.ValidateUtils;
import com.xiaojukeji.kafka.manager.common.zookeeper.znode.config.TopicQuotaData;
import com.xiaojukeji.kafka.manager.common.entity.pojo.TopicDO;
import com.xiaojukeji.kafka.manager.service.cache.LogicalClusterMetadataManager;
import com.xiaojukeji.kafka.manager.service.cache.PhysicalClusterMetadataManager;
import com.xiaojukeji.kafka.manager.service.service.gateway.AuthorityService;
import com.xiaojukeji.kafka.manager.service.service.gateway.QuotaService;
import com.xiaojukeji.kafka.manager.service.service.TopicManagerService;
import com.xiaojukeji.kafka.manager.service.strategy.AbstractAllocateQuotaStrategy;
@@ -28,6 +33,12 @@ public class QuotaServiceImpl implements QuotaService {
@Autowired
private AbstractAllocateQuotaStrategy allocateQuotaStrategy;
@Autowired
private LogicalClusterMetadataManager logicalClusterMetadataManager;
@Autowired
private AuthorityService authorityService;
@Override
public int addTopicQuota(TopicQuota topicQuotaDO) {
return KafkaZookeeperUtils.setTopicQuota(
@@ -78,4 +89,34 @@ public class QuotaServiceImpl implements QuotaService {
}
return Boolean.TRUE;
}
@Override
public ResultStatus addTopicQuotaByAuthority(TopicQuota topicQuota) {
// 获取物理集群id
Long physicalClusterId = logicalClusterMetadataManager.getPhysicalClusterId(topicQuota.getClusterId());
if (ValidateUtils.isNull(physicalClusterId)) {
return ResultStatus.CLUSTER_NOT_EXIST;
}
// 权限判断(access 0:无权限, 1:读, 2:写, 3:读写，4:可管理)
AuthorityDO authority = authorityService.getAuthority(physicalClusterId,
topicQuota.getTopicName(), topicQuota.getAppId());
if (ValidateUtils.isNull(authority) || authority.getAccess() == TopicAuthorityEnum.DENY.getCode()) {
return ResultStatus.USER_WITHOUT_AUTHORITY;
}
if (authority.getAccess() == TopicAuthorityEnum.READ.getCode()) {
// 可以消费
topicQuota.setProduceQuota(null);
}
if (authority.getAccess() == TopicAuthorityEnum.WRITE.getCode()) {
// 可以生产
topicQuota.setConsumeQuota(null);
}
// 设置物理集群id
topicQuota.setClusterId(physicalClusterId);
// 添加配额
if (addTopicQuota(topicQuota) > 0) {
return ResultStatus.SUCCESS;
}
return ResultStatus.ZOOKEEPER_WRITE_FAILED;
}
}

2
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/service/impl/BrokerServiceImpl.java
View File

@@ -91,7 +91,7 @@ public class BrokerServiceImpl implements BrokerService {
for (BrokerDO brokerDO : brokerDOList) {
PeakFlowStatusEnum peakFlowStatus = getPeakFlowStatus(brokerDO.getMaxAvgBytesIn(), peakFlow);
peakFlowStatusMap.put(
peakFlowStatus.code,
peakFlowStatus.getCode(),
peakFlowStatusMap.getOrDefault(peakFlowStatus.getCode(), 0) + 1
);
}

33
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/service/impl/TopicManagerServiceImpl.java
View File

@@ -20,6 +20,7 @@ import com.xiaojukeji.kafka.manager.common.entity.pojo.gateway.AuthorityDO;
import com.xiaojukeji.kafka.manager.common.utils.DateUtils;
import com.xiaojukeji.kafka.manager.common.utils.JsonUtils;
import com.xiaojukeji.kafka.manager.common.utils.NumberUtils;
import com.xiaojukeji.kafka.manager.common.utils.SpringTool;
import com.xiaojukeji.kafka.manager.common.utils.ValidateUtils;
import com.xiaojukeji.kafka.manager.common.zookeeper.znode.brokers.TopicMetadata;
import com.xiaojukeji.kafka.manager.common.zookeeper.znode.config.TopicQuotaData;
@@ -618,6 +619,38 @@ public class TopicManagerServiceImpl implements TopicManagerService {
return topicBusinessInfo;
}
@Override
public ResultStatus addAuthority(AuthorityDO authorityDO) {
// 查询该用户拥有的应用
List<AppDO> appDOs = appService.getByPrincipal(SpringTool.getUserName());
if (ValidateUtils.isEmptyList(appDOs)) {
// 该用户无应用，需要先申请应用
return ResultStatus.APP_NOT_EXIST;
}
List<Long> appIds = appDOs.stream().map(AppDO::getId).collect(Collectors.toList());
if (!appIds.contains(authorityDO.getAppId())) {
// 入参中的appId，该用户未拥有
return ResultStatus.APP_NOT_EXIST;
}
// 获取物理集群id
Long physicalClusterId = logicalClusterMetadataManager.getPhysicalClusterId(authorityDO.getClusterId());
if (ValidateUtils.isNull(physicalClusterId)) {
// 集群不存在
return ResultStatus.CLUSTER_NOT_EXIST;
}
TopicDO topic = getByTopicName(physicalClusterId, authorityDO.getTopicName());
if (ValidateUtils.isNull(topic)) {
// topic不存在
return ResultStatus.TOPIC_NOT_EXIST;
}
// 设置物理集群id
authorityDO.setClusterId(physicalClusterId);
if (authorityService.addAuthority(authorityDO) > 0) {
return ResultStatus.SUCCESS;
}
return ResultStatus.MYSQL_ERROR;
}
private RdTopicBasic convert2RdTopicBasic(ClusterDO clusterDO,
String topicName,
TopicDO topicDO,

41
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/service/impl/TopicServiceImpl.java
View File

@@ -223,6 +223,7 @@ public class TopicServiceImpl implements TopicService {
basicDTO.setCreateTime(topicMetadata.getCreateTime());
basicDTO.setModifyTime(topicMetadata.getModifyTime());
basicDTO.setRetentionTime(PhysicalClusterMetadataManager.getTopicRetentionTime(clusterId, topicName));
basicDTO.setRetentionBytes(PhysicalClusterMetadataManager.getTopicRetentionBytes(clusterId, topicName));
TopicDO topicDO = topicManagerService.getByTopicName(clusterId, topicName);
if (!ValidateUtils.isNull(topicDO)) {
@@ -648,10 +649,11 @@ public class TopicServiceImpl implements TopicService {
List<String> dataList = new ArrayList<>();
int currentSize = dataList.size();
while (dataList.size() < maxMsgNum) {
if (remainingWaitMs <= 0) {
break;
}
try {
if (remainingWaitMs <= 0) {
break;
}
ConsumerRecords<String, String> records = kafkaConsumer.poll(TopicSampleConstant.POLL_TIME_OUT_UNIT_MS);
for (ConsumerRecord record : records) {
String value = (String) record.value();
@@ -661,20 +663,22 @@ public class TopicServiceImpl implements TopicService {
: value
);
}
// 当前批次一条数据都没拉取到，则结束拉取
if (dataList.size() - currentSize == 0) {
break;
}
currentSize = dataList.size();
// 检查是否超时
long elapsed = System.currentTimeMillis() - begin;
if (elapsed >= maxWaitMs) {
break;
}
remainingWaitMs = maxWaitMs - elapsed;
} catch (Exception e) {
LOGGER.error("fetch topic data failed, TopicPartitions:{}.", kafkaConsumer.assignment(), e);
}
// 当前批次一条数据都没拉取到，则结束拉取
if (dataList.size() - currentSize == 0) {
break;
}
currentSize = dataList.size();
// 检查是否超时
long elapsed = System.currentTimeMillis() - begin;
if (elapsed >= maxWaitMs) {
break;
}
remainingWaitMs = maxWaitMs - elapsed;
}
return dataList.subList(0, Math.min(dataList.size(), maxMsgNum));
}
@@ -698,14 +702,15 @@ public class TopicServiceImpl implements TopicService {
: value
);
}
if (System.currentTimeMillis() - timestamp > timeout
|| dataList.size() >= maxMsgNum) {
break;
}
Thread.sleep(10);
} catch (Exception e) {
LOGGER.error("fetch topic data failed, TopicPartitions:{}.", kafkaConsumer.assignment(), e);
}
if (System.currentTimeMillis() - timestamp > timeout || dataList.size() >= maxMsgNum) {
// 超时或者是数据已采集足够时, 直接返回
break;
}
}
return dataList.subList(0, Math.min(dataList.size(), maxMsgNum));
}

2
kafka-manager-dao/src/main/resources/mapper/HeartbeatDao.xml
View File

@@ -11,7 +11,7 @@
</resultMap>
<insert id="replace" parameterType="com.xiaojukeji.kafka.manager.common.entity.pojo.HeartbeatDO">
REPLACE heartbeat (ip, hostname) VALUES (#{ip}, #{hostname})
REPLACE heartbeat (ip, hostname, modify_time) VALUES (#{ip}, #{hostname}, #{modifyTime})
</insert>
<select id="selectActiveHosts" parameterType="java.util.Date" resultMap="HeartbeatMap">

20
kafka-manager-extends/kafka-manager-account/src/main/java/com/xiaojukeji/kafka/manager/account/LoginService.java
View File

@@ -12,9 +12,29 @@ import javax.servlet.http.HttpServletResponse;
* @date 20/8/20
*/
public interface LoginService {
/**
* 登录
* @param request HttpServletRequest
* @param response HttpServletResponse
* @param dto 登录信息
* @return 登录结果
*/
Result<Account> login(HttpServletRequest request, HttpServletResponse response, LoginDTO dto);
/**
* 登出
* @param request HttpServletRequest
* @param response HttpServletResponse
* @param needJump2LoginPage 是否需要跳转到登录页
*/
void logout(HttpServletRequest request, HttpServletResponse response, Boolean needJump2LoginPage);
/**
* 检查是否登录
* @param request HttpServletRequest
* @param response HttpServletResponse
* @param classRequestMappingValue request-mapping的value
* @return 检查结果, false:未登录或无权限, true:已登录并且有权限
*/
boolean checkLogin(HttpServletRequest request, HttpServletResponse response, String classRequestMappingValue);
}

2
kafka-manager-extends/kafka-manager-account/src/main/java/com/xiaojukeji/kafka/manager/account/component/AbstractSingleSignOn.java
View File

@@ -7,7 +7,7 @@ import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
/**
* 单点登录抽象类
* 登录抽象类
* @author zengqiao
* @date 20/8/20
*/

50
kafka-manager-extends/kafka-manager-account/src/main/java/com/xiaojukeji/kafka/manager/account/component/login/trick/TrickLoginService.java Normal file
View File

@@ -0,0 +1,50 @@
package com.xiaojukeji.kafka.manager.account.component.login.trick;
import com.xiaojukeji.kafka.manager.common.constant.TrickLoginConstant;
import com.xiaojukeji.kafka.manager.service.service.ConfigService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import javax.servlet.http.HttpServletRequest;
import java.util.Optional;
/**
* @author zengqiao
* @date 21/5/18
*/
@Service
public class TrickLoginService {
private final static Logger LOGGER = LoggerFactory.getLogger(TrickLoginService.class);
@Autowired
private ConfigService configService;
/**
* 是否开启trick的方式登录
*/
public boolean isTrickLoginOn(HttpServletRequest request) {
return TrickLoginConstant.TRICK_LOGIN_SWITCH_ON.equals(request.getHeader(TrickLoginConstant.TRICK_LOGIN_SWITCH));
}
/**
* 开启trick方式登录后，当前用户是否可以登录
*/
public String checkTrickLogin(HttpServletRequest request) {
String trickLoginUser = request.getHeader(TrickLoginConstant.TRICK_LOGIN_USER);
LOGGER.info("class=TrickLoginService||method=checkTrickLogin||user={}||uri={}||msg=try trick login", trickLoginUser, request.getRequestURI());
if (!checkTrickLogin(trickLoginUser)) {
LOGGER.warn("class=TrickLoginService||method=checkTrickLogin||user={}||uri={}||msg=trick login failed", trickLoginUser, request.getRequestURI());
return null;
}
return trickLoginUser;
}
private boolean checkTrickLogin(String trickLoginUser) {
return Optional.ofNullable(configService.getArrayByKey(TrickLoginConstant.TRICK_LOGIN_LEGAL_USER_CONFIG_KEY, String.class))
.filter(names -> names.contains(trickLoginUser))
.isPresent();
}
}

6
kafka-manager-extends/kafka-manager-account/src/main/java/com/xiaojukeji/kafka/manager/account/impl/LoginServiceImpl.java
View File

@@ -3,6 +3,7 @@ package com.xiaojukeji.kafka.manager.account.impl;
import com.xiaojukeji.kafka.manager.account.AccountService;
import com.xiaojukeji.kafka.manager.account.component.AbstractSingleSignOn;
import com.xiaojukeji.kafka.manager.account.LoginService;
import com.xiaojukeji.kafka.manager.account.component.login.trick.TrickLoginService;
import com.xiaojukeji.kafka.manager.common.bizenum.AccountRoleEnum;
import com.xiaojukeji.kafka.manager.common.constant.ApiPrefix;
import com.xiaojukeji.kafka.manager.common.constant.LoginConstant;
@@ -31,6 +32,9 @@ public class LoginServiceImpl implements LoginService {
@Autowired
private AccountService accountService;
@Autowired
private TrickLoginService trickLoginService;
@Autowired
private AbstractSingleSignOn singleSignOn;
@@ -80,7 +84,7 @@ public class LoginServiceImpl implements LoginService {
return true;
}
String username = singleSignOn.checkLoginAndGetLdap(request);
String username = trickLoginService.isTrickLoginOn(request)? trickLoginService.checkTrickLogin(request): singleSignOn.checkLoginAndGetLdap(request);
if (ValidateUtils.isBlank(username)) {
// 未登录, 则返回false, 同时重定向到登录页面
singleSignOn.setRedirectToLoginPage(response);

39
kafka-manager-extends/kafka-manager-openapi/src/main/java/com/xiaojukeji/kafka/manager/openapi/common/dto/TopicAuthorityDTO.java Normal file
View File

@@ -0,0 +1,39 @@
package com.xiaojukeji.kafka.manager.openapi.common.dto;
import com.xiaojukeji.kafka.manager.common.entity.dto.ClusterTopicDTO;
import com.xiaojukeji.kafka.manager.common.utils.ValidateUtils;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
@ApiModel(description = "权限调整")
public class TopicAuthorityDTO extends ClusterTopicDTO {
@ApiModelProperty(value = "appId")
private String appId;
@ApiModelProperty(value = "0:无权限, 1:读, 2:写, 3:读写, 4:可管理")
private Integer access;
public String getAppId() {
return appId;
}
public void setAppId(String appId) {
this.appId = appId;
}
public Integer getAccess() {
return access;
}
public void setAccess(Integer access) {
this.access = access;
}
@Override
public boolean paramLegal() {
return !ValidateUtils.isNullOrLessThanZero(clusterId)
&& !ValidateUtils.isBlank(topicName)
&& !ValidateUtils.isBlank(appId)
&& !ValidateUtils.isNullOrLessThanZero(access);
}
}

14
kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/component/AbstractScheduledTask.java
View File

@@ -148,10 +148,12 @@ public abstract class AbstractScheduledTask<E extends Comparable> implements Sch
}
Collections.sort(allTaskList);
List<HeartbeatDO> hostList = heartbeatDao.selectActiveHosts(
new Date(now - ScheduledTaskConstant.HEARTBEAT_TIME)
);
List<HeartbeatDO> hostList = heartbeatDao.selectActiveHosts(new Date(now - ScheduledTaskConstant.HEARTBEAT_TIME));
if (ValidateUtils.isEmptyList(hostList)) {
// 当前无机器注册，导致周期任务(Topic指标存DB等任务)不可被触发执行。
// 大概率原因可能是：DB的时区不对，注册的时间错误导致查询不出来。
// 如果是单台方式部署的Logi-KM，那么也可能是服务新上线，或者是服务不正常导致的。
LOGGER.error("customScheduled task running, but without registrant, and so scheduled tasks can't execute, scheduledName:{}.", scheduledName);
return Lists.newArrayList();
}
@@ -163,8 +165,10 @@ public abstract class AbstractScheduledTask<E extends Comparable> implements Sch
idx++;
}
if (idx == hostList.size()) {
// 当前机器未注册
LOGGER.error("customScheduled task running, registers not conclude, scheduledName:{}.", scheduledName);
// 当前机器未注册, 原因可能是：
// 1、当前服务新上线，确实暂未注册到DB中。
// 2、当前服务异常，比如进行FGC等，导致注册任务停止了。
LOGGER.warn("customScheduled task running, registrants not conclude present machine, scheduledName:{}.", scheduledName);
return Lists.newArrayList();
}

3
kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/component/Heartbeat.java
View File

@@ -10,6 +10,8 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.util.Date;
/**
* @author limeng
* @date 20/8/10
@@ -30,6 +32,7 @@ public class Heartbeat {
HeartbeatDO heartbeatDO = new HeartbeatDO();
heartbeatDO.setIp(NetUtils.localIp());
heartbeatDO.setHostname(NetUtils.localHostname());
heartbeatDO.setModifyTime(new Date());
heartbeatDao.replace(heartbeatDO);
} catch (Exception e) {
LOGGER.error("flush heartbeat failed.", e);

19
kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/schedule/metadata/FlushTopicRetentionTime.java → kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/schedule/metadata/FlushTopicProperties.java
View File

@@ -14,13 +14,14 @@ import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.util.List;
import java.util.Properties;
/**
* @author zengqiao
* @date 20/7/23
*/
@Component
public class FlushTopicRetentionTime {
public class FlushTopicProperties {
private final static Logger LOGGER = LoggerFactory.getLogger(LogConstant.SCHEDULED_TASK_LOGGER);
@Autowired
@@ -33,7 +34,7 @@ public class FlushTopicRetentionTime {
try {
flush(clusterDO);
} catch (Exception e) {
LOGGER.error("flush topic retention time failed, clusterId:{}.", clusterDO.getId(), e);
LOGGER.error("flush topic properties failed, clusterId:{}.", clusterDO.getId(), e);
}
}
}
@@ -41,22 +42,20 @@ public class FlushTopicRetentionTime {
private void flush(ClusterDO clusterDO) {
ZkConfigImpl zkConfig = PhysicalClusterMetadataManager.getZKConfig(clusterDO.getId());
if (ValidateUtils.isNull(zkConfig)) {
LOGGER.error("flush topic retention time, get zk config failed, clusterId:{}.", clusterDO.getId());
LOGGER.error("flush topic properties, get zk config failed, clusterId:{}.", clusterDO.getId());
return;
}
for (String topicName: PhysicalClusterMetadataManager.getTopicNameList(clusterDO.getId())) {
try {
Long retentionTime = KafkaZookeeperUtils.getTopicRetentionTime(zkConfig, topicName);
if (retentionTime == null) {
LOGGER.warn("get topic retentionTime failed, clusterId:{} topicName:{}.",
clusterDO.getId(), topicName);
Properties properties = KafkaZookeeperUtils.getTopicProperties(zkConfig, topicName);
if (ValidateUtils.isNull(properties)) {
LOGGER.warn("get topic properties failed, clusterId:{} topicName:{}.", clusterDO.getId(), topicName);
continue;
}
PhysicalClusterMetadataManager.putTopicRetentionTime(clusterDO.getId(), topicName, retentionTime);
PhysicalClusterMetadataManager.putTopicProperties(clusterDO.getId(), topicName, properties);
} catch (Exception e) {
LOGGER.error("get topic retentionTime failed, clusterId:{} topicName:{}.",
clusterDO.getId(), topicName, e);
LOGGER.error("get topic properties failed, clusterId:{} topicName:{}.", clusterDO.getId(), topicName, e);
}
}
}

2
kafka-manager-web/pom.xml
View File

@@ -19,7 +19,7 @@
<springframework.boot.version>2.1.1.RELEASE</springframework.boot.version>
<spring-version>5.1.3.RELEASE</spring-version>
<failOnMissingWebXml>false</failOnMissingWebXml>
<tomcat.version>8.5.37</tomcat.version>
<tomcat.version>8.5.66</tomcat.version>
</properties>
<dependencies>

7
kafka-manager-web/src/main/java/com/xiaojukeji/kafka/manager/web/api/versionone/normal/NormalTopicController.java
View File

@@ -61,10 +61,7 @@ public class NormalTopicController {
@ApiOperation(value = "Topic基本信息", notes = "")
@RequestMapping(value = "{clusterId}/topics/{topicName}/basic-info", method = RequestMethod.GET)
@ResponseBody
public Result<TopicBasicVO> getTopicBasic(
@PathVariable Long clusterId,
@PathVariable String topicName,
@RequestParam(value = "isPhysicalClusterId", required = false) Boolean isPhysicalClusterId) {
public Result<TopicBasicVO> getTopicBasic(@PathVariable Long clusterId, @PathVariable String topicName, @RequestParam(value = "isPhysicalClusterId", required = false) Boolean isPhysicalClusterId) {
Long physicalClusterId = logicalClusterMetadataManager.getPhysicalClusterId(clusterId, isPhysicalClusterId);
if (ValidateUtils.isNull(physicalClusterId)) {
return Result.buildFrom(ResultStatus.CLUSTER_NOT_EXIST);
@@ -337,7 +334,7 @@ public class NormalTopicController {
}
return new Result<>(TopicModelConverter.convert2TopicMineAppVOList(
topicManagerService.getTopicMineApps(physicalClusterId, topicName, SpringTool.getUserName()))
topicManagerService.getTopicMineApps(physicalClusterId, topicName, SpringTool.getUserName()))
);
}

35
kafka-manager-web/src/main/java/com/xiaojukeji/kafka/manager/web/api/versionone/op/OpAuthorityController.java Normal file
View File

@@ -0,0 +1,35 @@
package com.xiaojukeji.kafka.manager.web.api.versionone.op;
import com.xiaojukeji.kafka.manager.common.entity.Result;
import com.xiaojukeji.kafka.manager.common.entity.ResultStatus;
import com.xiaojukeji.kafka.manager.common.utils.ValidateUtils;
import com.xiaojukeji.kafka.manager.openapi.common.dto.TopicAuthorityDTO;
import com.xiaojukeji.kafka.manager.service.service.TopicManagerService;
import com.xiaojukeji.kafka.manager.web.converters.AuthorityConverter;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
/**
* Authority操作相关接口
* @author zengqiao
* @date 21/5/18
*/
@Api(tags = "OP-Authority操作相关接口(REST)")
@RestController
public class OpAuthorityController {
@Autowired
private TopicManagerService topicManagerService;
@ApiOperation(value = "权限调整",notes = "权限调整")
@PostMapping(value = "topic-authorities")
@ResponseBody
public Result addAuthority(@RequestBody TopicAuthorityDTO dto) {
//非空校验
if (ValidateUtils.isNull(dto) || !dto.paramLegal()) {
return Result.buildFrom(ResultStatus.PARAM_ILLEGAL);
}
return Result.buildFrom(topicManagerService.addAuthority(AuthorityConverter.convert2AuthorityDO(dto)));
}
}

29
kafka-manager-web/src/main/java/com/xiaojukeji/kafka/manager/web/api/versionone/thirdpart/ThirdPartOpUtilController.java → kafka-manager-web/src/main/java/com/xiaojukeji/kafka/manager/web/api/versionone/op/OpLeaderController.java
View File

@@ -1,11 +1,13 @@
package com.xiaojukeji.kafka.manager.web.api.versionone.thirdpart;
package com.xiaojukeji.kafka.manager.web.api.versionone.op;
import com.xiaojukeji.kafka.manager.common.bizenum.RebalanceDimensionEnum;
import com.xiaojukeji.kafka.manager.common.bizenum.TaskStatusEnum;
import com.xiaojukeji.kafka.manager.common.constant.ApiPrefix;
import com.xiaojukeji.kafka.manager.common.entity.Result;
import com.xiaojukeji.kafka.manager.common.entity.ResultStatus;
import com.xiaojukeji.kafka.manager.common.entity.dto.op.RebalanceDTO;
import com.xiaojukeji.kafka.manager.common.entity.pojo.ClusterDO;
import com.xiaojukeji.kafka.manager.common.utils.JsonUtils;
import com.xiaojukeji.kafka.manager.common.utils.SpringTool;
import com.xiaojukeji.kafka.manager.common.utils.ValidateUtils;
import com.xiaojukeji.kafka.manager.service.service.AdminService;
@@ -16,22 +18,35 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
/**
* Leader操作[选举|切换]相关接口
* @author zengqiao
* @date 20/9/23
* @date 21/5/18
*/
@Api(tags = "开放接口-OP相关接口(REST)")
@Api(tags = "OP-Leader操作相关接口(REST)")
@RestController
@RequestMapping(ApiPrefix.API_V1_THIRD_PART_OP_PREFIX)
public class ThirdPartOpUtilController {
@RequestMapping(ApiPrefix.API_V1_OP_PREFIX)
public class OpLeaderController {
@Autowired
private AdminService adminService;
@Autowired
private ClusterService clusterService;
@ApiOperation(value = "优先副本选举状态")
@RequestMapping(value = {"leaders/preferred-replica-election-status", "utils/rebalance-status"}, method = RequestMethod.GET)
@ResponseBody
public Result preferredReplicaElectStatus(@RequestParam("clusterId") Long clusterId) {
ClusterDO clusterDO = clusterService.getById(clusterId);
if (ValidateUtils.isNull(clusterDO)) {
return Result.buildFrom(ResultStatus.CLUSTER_NOT_EXIST);
}
TaskStatusEnum statusEnum = adminService.preferredReplicaElectionStatus(clusterDO);
return new Result<>(JsonUtils.toJson(statusEnum));
}
@ApiOperation(value = "优先副本选举")
@RequestMapping(value = "op/rebalance", method = RequestMethod.POST)
@RequestMapping(value = {"leaders/preferred-replica-election", "utils/rebalance"}, method = RequestMethod.POST)
@ResponseBody
public Result preferredReplicaElect(@RequestBody RebalanceDTO reqObj) {
if (!reqObj.paramLegal()) {

37
kafka-manager-web/src/main/java/com/xiaojukeji/kafka/manager/web/api/versionone/op/OpQuotaController.java Normal file
View File

@@ -0,0 +1,37 @@
package com.xiaojukeji.kafka.manager.web.api.versionone.op;
import com.xiaojukeji.kafka.manager.common.constant.ApiPrefix;
import com.xiaojukeji.kafka.manager.common.entity.Result;
import com.xiaojukeji.kafka.manager.common.entity.ResultStatus;
import com.xiaojukeji.kafka.manager.common.entity.ao.gateway.TopicQuota;
import com.xiaojukeji.kafka.manager.common.entity.dto.gateway.TopicQuotaDTO;
import com.xiaojukeji.kafka.manager.common.utils.ValidateUtils;
import com.xiaojukeji.kafka.manager.service.service.gateway.QuotaService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
/**
* Quota操作相关接口
* @author zengqiao
* @date 21/5/18
*/
@Api(tags = "OP-Quota操作相关接口(REST)")
@RestController
@RequestMapping(ApiPrefix.API_V1_OP_PREFIX)
public class OpQuotaController {
@Autowired
private QuotaService quotaService;
@ApiOperation(value = "配额调整",notes = "配额调整")
@RequestMapping(value = "topic-quotas",method = RequestMethod.POST)
@ResponseBody
public Result addTopicQuota(@RequestBody TopicQuotaDTO dto) {
if (ValidateUtils.isNull(dto) || !dto.paramLegal()) {
// 非空校验
return Result.buildFrom(ResultStatus.PARAM_ILLEGAL);
}
return Result.buildFrom(quotaService.addTopicQuotaByAuthority(TopicQuota.buildFrom(dto)));
}
}

194
kafka-manager-web/src/main/java/com/xiaojukeji/kafka/manager/web/api/versionone/op/OpUtilsController.java → kafka-manager-web/src/main/java/com/xiaojukeji/kafka/manager/web/api/versionone/op/OpTopicController.java
View File

@@ -1,24 +1,23 @@
package com.xiaojukeji.kafka.manager.web.api.versionone.op;
import com.xiaojukeji.kafka.manager.common.bizenum.RebalanceDimensionEnum;
import com.xiaojukeji.kafka.manager.common.bizenum.TaskStatusEnum;
import com.xiaojukeji.kafka.manager.common.constant.ApiPrefix;
import com.xiaojukeji.kafka.manager.common.constant.Constant;
import com.xiaojukeji.kafka.manager.common.constant.KafkaConstant;
import com.xiaojukeji.kafka.manager.common.entity.Result;
import com.xiaojukeji.kafka.manager.common.entity.ResultStatus;
import com.xiaojukeji.kafka.manager.common.entity.TopicOperationResult;
import com.xiaojukeji.kafka.manager.common.entity.dto.ClusterTopicDTO;
import com.xiaojukeji.kafka.manager.common.entity.dto.op.RebalanceDTO;
import com.xiaojukeji.kafka.manager.common.entity.dto.op.topic.*;
import com.xiaojukeji.kafka.manager.common.entity.dto.op.topic.TopicCreationDTO;
import com.xiaojukeji.kafka.manager.common.entity.dto.op.topic.TopicDeletionDTO;
import com.xiaojukeji.kafka.manager.common.entity.dto.op.topic.TopicExpansionDTO;
import com.xiaojukeji.kafka.manager.common.entity.dto.op.topic.TopicModificationDTO;
import com.xiaojukeji.kafka.manager.common.entity.pojo.ClusterDO;
import com.xiaojukeji.kafka.manager.common.entity.pojo.TopicDO;
import com.xiaojukeji.kafka.manager.common.utils.SpringTool;
import com.xiaojukeji.kafka.manager.common.utils.ValidateUtils;
import com.xiaojukeji.kafka.manager.service.service.AdminService;
import com.xiaojukeji.kafka.manager.service.service.ClusterService;
import com.xiaojukeji.kafka.manager.common.utils.JsonUtils;
import com.xiaojukeji.kafka.manager.common.utils.SpringTool;
import com.xiaojukeji.kafka.manager.service.service.TopicManagerService;
import com.xiaojukeji.kafka.manager.common.constant.ApiPrefix;
import com.xiaojukeji.kafka.manager.common.entity.TopicOperationResult;
import com.xiaojukeji.kafka.manager.service.utils.TopicCommands;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
@@ -30,25 +29,25 @@ import java.util.List;
import java.util.Properties;
/**
* 运维工具类
* Topic操作相关接口
* @author zengqiao
* @date 20/4/2
* @date 21/5/18
*/
@Api(tags = "OP-Utils相关接口(REST)")
@Api(tags = "OP-Topic操作相关接口(REST)")
@RestController
@RequestMapping(ApiPrefix.API_V1_OP_PREFIX)
public class OpUtilsController {
@Autowired
private ClusterService clusterService;
public class OpTopicController {
@Autowired
private AdminService adminService;
@Autowired
private ClusterService clusterService;
@Autowired
private TopicManagerService topicManagerService;
@ApiOperation(value = "创建Topic")
@RequestMapping(value = {"utils/topics"}, method = RequestMethod.POST)
@RequestMapping(value = {"topics", "utils/topics"}, method = RequestMethod.POST)
@ResponseBody
public Result createCommonTopic(@RequestBody TopicCreationDTO dto) {
Result<ClusterDO> rc = checkParamAndGetClusterDO(dto);
@@ -76,8 +75,66 @@ public class OpUtilsController {
return Result.buildFrom(rs);
}
@ApiOperation(value = "Topic删除", notes = "单次不允许超过10个Topic")
@RequestMapping(value = {"topics", "utils/topics"}, method = RequestMethod.DELETE)
@ResponseBody
public Result<List<TopicOperationResult>> deleteTopics(@RequestBody List<TopicDeletionDTO> dtoList) {
if (ValidateUtils.isNull(dtoList) || dtoList.size() > Constant.MAX_TOPIC_OPERATION_SIZE_PER_REQUEST) {
return Result.buildFrom(ResultStatus.PARAM_ILLEGAL);
}
String operator = SpringTool.getUserName();
List<TopicOperationResult> resultList = new ArrayList<>();
for (TopicDeletionDTO dto: dtoList) {
Result<ClusterDO> rc = checkParamAndGetClusterDO(dto);
if (rc.getCode() != ResultStatus.SUCCESS.getCode()) {
resultList.add(TopicOperationResult.buildFrom(dto.getClusterId(), dto.getTopicName(), rc));
continue;
}
// 参数检查合法, 开始删除Topic
ResultStatus statusEnum = adminService.deleteTopic(rc.getData(), dto.getTopicName(), operator);
resultList.add(TopicOperationResult.buildFrom(dto.getClusterId(), dto.getTopicName(), statusEnum));
}
for (TopicOperationResult operationResult: resultList) {
if (!Constant.SUCCESS.equals(operationResult.getCode())) {
return Result.buildFrom(ResultStatus.OPERATION_FAILED, resultList);
}
}
return new Result<>(resultList);
}
@ApiOperation(value = "修改Topic", notes = "")
@RequestMapping(value = {"topics", "utils/topics"}, method = RequestMethod.PUT)
@ResponseBody
public Result modifyTopic(@RequestBody TopicModificationDTO dto) {
Result<ClusterDO> rc = checkParamAndGetClusterDO(dto);
if (rc.getCode() != ResultStatus.SUCCESS.getCode()) {
return rc;
}
ClusterDO clusterDO = rc.getData();
// 获取属性
Properties properties = dto.getProperties();
if (ValidateUtils.isNull(properties)) {
properties = new Properties();
}
properties.put(KafkaConstant.RETENTION_MS_KEY, String.valueOf(dto.getRetentionTime()));
// 操作修改
String operator = SpringTool.getUserName();
ResultStatus rs = TopicCommands.modifyTopicConfig(clusterDO, dto.getTopicName(), properties);
if (!ResultStatus.SUCCESS.equals(rs)) {
return Result.buildFrom(rs);
}
topicManagerService.modifyTopicByOp(dto.getClusterId(), dto.getTopicName(), dto.getAppId(), dto.getDescription(), operator);
return new Result();
}
@ApiOperation(value = "Topic扩分区", notes = "")
@RequestMapping(value = {"utils/expand-partitions"}, method = RequestMethod.PUT)
@RequestMapping(value = {"topics/expand-partitions", "utils/expand-partitions"}, method = RequestMethod.PUT)
@ResponseBody
public Result<List<TopicOperationResult>> expandTopics(@RequestBody List<TopicExpansionDTO> dtoList) {
if (ValidateUtils.isNull(dtoList) || dtoList.size() > Constant.MAX_TOPIC_OPERATION_SIZE_PER_REQUEST) {
@@ -112,108 +169,6 @@ public class OpUtilsController {
return new Result<>(resultList);
}
@ApiOperation(value = "Topic删除", notes = "单次不允许超过10个Topic")
@RequestMapping(value = {"utils/topics"}, method = RequestMethod.DELETE)
@ResponseBody
public Result<List<TopicOperationResult>> deleteTopics(@RequestBody List<TopicDeletionDTO> dtoList) {
if (ValidateUtils.isNull(dtoList) || dtoList.size() > Constant.MAX_TOPIC_OPERATION_SIZE_PER_REQUEST) {
return Result.buildFrom(ResultStatus.PARAM_ILLEGAL);
}
String operator = SpringTool.getUserName();
List<TopicOperationResult> resultList = new ArrayList<>();
for (TopicDeletionDTO dto: dtoList) {
Result<ClusterDO> rc = checkParamAndGetClusterDO(dto);
if (rc.getCode() != ResultStatus.SUCCESS.getCode()) {
resultList.add(TopicOperationResult.buildFrom(dto.getClusterId(), dto.getTopicName(), rc));
continue;
}
// 参数检查合法, 开始删除Topic
ResultStatus statusEnum = adminService.deleteTopic(rc.getData(), dto.getTopicName(), operator);
resultList.add(TopicOperationResult.buildFrom(dto.getClusterId(), dto.getTopicName(), statusEnum));
}
for (TopicOperationResult operationResult: resultList) {
if (!Constant.SUCCESS.equals(operationResult.getCode())) {
return Result.buildFrom(ResultStatus.OPERATION_FAILED, resultList);
}
}
return new Result<>(resultList);
}
@ApiOperation(value = "修改Topic", notes = "")
@RequestMapping(value = {"utils/topics"}, method = RequestMethod.PUT)
@ResponseBody
public Result modifyTopic(@RequestBody TopicModificationDTO dto) {
Result<ClusterDO> rc = checkParamAndGetClusterDO(dto);
if (rc.getCode() != ResultStatus.SUCCESS.getCode()) {
return rc;
}
ClusterDO clusterDO = rc.getData();
// 获取属性
Properties properties = dto.getProperties();
if (ValidateUtils.isNull(properties)) {
properties = new Properties();
}
properties.put(KafkaConstant.RETENTION_MS_KEY, String.valueOf(dto.getRetentionTime()));
// 操作修改
String operator = SpringTool.getUserName();
ResultStatus rs = TopicCommands.modifyTopicConfig(clusterDO, dto.getTopicName(), properties);
if (!ResultStatus.SUCCESS.equals(rs)) {
return Result.buildFrom(rs);
}
topicManagerService.modifyTopicByOp(dto.getClusterId(), dto.getTopicName(), dto.getAppId(), dto.getDescription(), operator);
return new Result();
}
@ApiOperation(value = "优先副本选举状态")
@RequestMapping(value = "utils/rebalance-status", method = RequestMethod.GET)
@ResponseBody
public Result preferredReplicaElectStatus(@RequestParam("clusterId") Long clusterId) {
ClusterDO clusterDO = clusterService.getById(clusterId);
if (ValidateUtils.isNull(clusterDO)) {
return Result.buildFrom(ResultStatus.CLUSTER_NOT_EXIST);
}
TaskStatusEnum statusEnum = adminService.preferredReplicaElectionStatus(clusterDO);
return new Result<>(JsonUtils.toJson(statusEnum));
}
@ApiOperation(value = "优先副本选举")
@RequestMapping(value = "utils/rebalance", method = RequestMethod.POST)
@ResponseBody
public Result preferredReplicaElect(@RequestBody RebalanceDTO reqObj) {
if (!reqObj.paramLegal()) {
return Result.buildFrom(ResultStatus.PARAM_ILLEGAL);
}
ClusterDO clusterDO = clusterService.getById(reqObj.getClusterId());
if (ValidateUtils.isNull(clusterDO)) {
return Result.buildFrom(ResultStatus.CLUSTER_NOT_EXIST);
}
ResultStatus rs = null;
if (RebalanceDimensionEnum.CLUSTER.getCode().equals(reqObj.getDimension())) {
// 按照Cluster纬度均衡
rs = adminService.preferredReplicaElection(clusterDO, SpringTool.getUserName());
} else if (RebalanceDimensionEnum.BROKER.getCode().equals(reqObj.getDimension())) {
// 按照Broker纬度均衡
rs = adminService.preferredReplicaElection(clusterDO, reqObj.getBrokerId(), SpringTool.getUserName());
} else if (RebalanceDimensionEnum.TOPIC.getCode().equals(reqObj.getDimension())) {
// 按照Topic纬度均衡
rs = adminService.preferredReplicaElection(clusterDO, reqObj.getTopicName(), SpringTool.getUserName());
} else if (RebalanceDimensionEnum.PARTITION.getCode().equals(reqObj.getDimension())) {
// 按照Partition纬度均衡
rs = adminService.preferredReplicaElection(clusterDO, reqObj.getTopicName(), reqObj.getPartitionId(), SpringTool.getUserName());
} else {
return Result.buildFrom(ResultStatus.PARAM_ILLEGAL);
}
return Result.buildFrom(rs);
}
private Result<ClusterDO> checkParamAndGetClusterDO(ClusterTopicDTO dto) {
if (!dto.paramLegal()) {
return Result.buildFrom(ResultStatus.PARAM_ILLEGAL);
@@ -226,4 +181,3 @@ public class OpUtilsController {
return new Result<>(clusterDO);
}
}

2
kafka-manager-web/src/main/java/com/xiaojukeji/kafka/manager/web/api/versionone/thirdpart/ThirdPartTopicController.java
View File

@@ -133,4 +133,4 @@ public class ThirdPartTopicController {
topicManagerService.getTopicAuthorizedApps(physicalClusterId, topicName))
);
}
}
}

15
kafka-manager-web/src/main/java/com/xiaojukeji/kafka/manager/web/converters/AuthorityConverter.java Normal file
View File

@@ -0,0 +1,15 @@
package com.xiaojukeji.kafka.manager.web.converters;
import com.xiaojukeji.kafka.manager.common.entity.pojo.gateway.AuthorityDO;
import com.xiaojukeji.kafka.manager.openapi.common.dto.TopicAuthorityDTO;
public class AuthorityConverter {
public static AuthorityDO convert2AuthorityDO(TopicAuthorityDTO dto) {
AuthorityDO authorityDO = new AuthorityDO();
authorityDO.setAppId(dto.getAppId());
authorityDO.setClusterId(dto.getClusterId());
authorityDO.setTopicName(dto.getTopicName());
authorityDO.setAccess(dto.getAccess());
return authorityDO;
}
}

1
kafka-manager-web/src/main/java/com/xiaojukeji/kafka/manager/web/converters/TopicModelConverter.java
View File

@@ -31,6 +31,7 @@ public class TopicModelConverter {
vo.setReplicaNum(dto.getReplicaNum());
vo.setPrincipals(dto.getPrincipals());
vo.setRetentionTime(dto.getRetentionTime());
vo.setRetentionBytes(dto.getRetentionBytes());
vo.setCreateTime(dto.getCreateTime());
vo.setModifyTime(dto.getModifyTime());
vo.setScore(dto.getScore());

6
kafka-manager-web/src/main/resources/application.yml
View File

@@ -84,11 +84,11 @@ monitor:
nid: 2
user-token: 1234567890
mon:
base-url: http://127.0.0.1:8032
base-url: http://127.0.0.1:8000 # 夜莺v4版本，默认端口统一调整为了8000
sink:
base-url: http://127.0.0.1:8006
base-url: http://127.0.0.1:8000 # 夜莺v4版本，默认端口统一调整为了8000
rdb:
base-url: http://127.0.0.1:80
base-url: http://127.0.0.1:8000 # 夜莺v4版本，默认端口统一调整为了8000
notify:
kafka:

8
pom.xml
View File

@@ -16,7 +16,7 @@
</parent>
<properties>
<kafka-manager.revision>2.4.0-SNAPSHOT</kafka-manager.revision>
<kafka-manager.revision>2.4.2-SNAPSHOT</kafka-manager.revision>
<swagger2.version>2.7.0</swagger2.version>
<swagger.version>1.5.13</swagger.version>
@@ -26,7 +26,7 @@
<java_target_version>1.8</java_target_version>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<file_encoding>UTF-8</file_encoding>
<tomcat.version>8.5.37</tomcat.version>
<tomcat.version>8.5.66</tomcat.version>
</properties>
<modules>
@@ -147,7 +147,7 @@
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<version>2.9.10.5</version>
<version>2.9.10.8</version>
</dependency>
<!-- commons -->
@@ -215,7 +215,7 @@
<dependency>
<groupId>commons-beanutils</groupId>
<artifactId>commons-beanutils</artifactId>
<version>1.9.3</version>
<version>1.9.4</version>
</dependency>
<dependency>