bump version to v2.4.2 and add release notes

bump tomcat version to 8.5.66

README.md

@@ -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建立了基于分角色、多场景视角的体验地图。分别是：**用户体验地图、运维体验地图、运营体验地图**
@@ -66,6 +67,7 @@
 - [滴滴Logi-KafkaManager 开源之路](https://xie.infoq.cn/article/0223091a99e697412073c0d64)
 - [滴滴Logi-KafkaManager 系列视频教程](https://mp.weixin.qq.com/s/9X7gH0tptHPtfjPPSdGO8g)
 - [kafka实践（十五）：滴滴开源Kafka管控平台 Logi-KafkaManager研究--A叶子叶来](https://blog.csdn.net/yezonggang/article/details/113106244)
+- [kafka的灵魂伴侣Logi-KafkaManager系列文章专栏 --石臻](https://blog.csdn.net/u010634066/category_10977588.html)
 
 ## 3 滴滴Logi开源用户交流群
 

Releases_Notes.md

@@ -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

build.sh

@@ -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}
 

docs/dev_guide/dynamic_config_manager.md

@@ -119,4 +119,24 @@ TOPIC_INSUFFICIENT_PARTITION_CONFIG
     50
   ]
 }
-```
+```
+## 4、专家服务——Topic资源治理
+
+首先，我们认为在一定的时间长度内，Topic的分区offset没有任何变化的Topic，即没有数据写入的Topic，为过期的Topic。
+
+Topic分区不足相关的动态配置(页面在运维管控->平台管理->配置管理)：
+
+配置Key：
+```
+EXPIRED_TOPIC_CONFIG
+```
+
+配置Value：
+```json
+{
+  "minExpiredDay": 30,  #过期时间大于此值才显示
+  "ignoreClusterIdList": [  # 忽略的集群
+    50
+  ]
+}
+```

docs/install_guide/create_mysql_table.sql

@@ -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`, `description`) values('SD_QUEUE_SIZE', 'SD_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`, `description`) values('SD_APP_RATE', 'SD_APP_RATE', 100000000, 1, '任意一个App限速');
-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`, `description`) values('SD_IP_RATE', 'SD_IP_RATE', 100000000, 1, '任意一个IP限速');
-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`, `description`) values('SD_SP_RATE', 'app_01234567', 100000000, 1, '指定App限速');
-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_SP_RATE', '192.168.0.1', 100000000, 1, '指定IP限速');
 
 --
 -- Table structure for table `heartbeat`

docs/user_guide/call_api_bypass_login.md

@@ -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"]
+```
+
+ 
+
+步骤三：解释说明
+
+设置完成上面两步之后，就可以直接调用需要登录的接口了。
+
+但是还有一点需要注意，绕过的用户仅能调用他有权限的接口，比如一个普通用户，那么他就只能调用普通的接口，不能去调用运维人员的接口。
+

docs/user_guide/faq.md

@@ -20,8 +20,14 @@
 - 8、`topic biz data not exist`错误及处理方式
 - 9、进程启动后，如何查看API文档
 - 10、如何创建告警组？
-- 11、连接信息、耗时信息为什么没有数据？
+- 11、连接信息、耗时信息、磁盘信息为什么没有数据？
 - 12、逻辑集群申请审批通过之后为什么看不到逻辑集群？
+- 13、heartbeat表关联业务和使用场景是什么？
+- 14、集群的删除，是否会真正的删除集群？
+- 15、APP(应用)如何被使用起来？
+- 16、为什么下线应用提示operation forbidden？
+- 17、删除Topic成功，为什么过一会儿之后又出现了？
+- 18、如何在不登录的情况下，调用一些需要登录的接口？
 
 ---
 
@@ -70,7 +76,7 @@
 
 - 3、数据库时区问题。
 
-检查MySQL的topic_metrics表，查看是否有数据，如果有数据，那么再检查设置的时区是否正确。
+检查MySQL的cluster表的gmt_modify字段，做一个update动作，看这个gmt_modify时间是否是当前时间，如果不是，那么就是时区问题了。时区不对问题具体可以搜索一下看如何解决。
 
 ---
 
@@ -113,11 +119,14 @@
 
 ### 10、如何创建告警组？
 
-这块需要配合监控系统进行使用，现在默认已经实现了夜莺的对接，当然也可以对接自己内部的监控系统，不过需要实现一些接口。
+告警组的创建需要到Logi-KM对接的监控系统中创建，比如我们现在默认是对接了夜莺，那么告警组需要到夜莺中创建，如果没有安装夜莺，那么需要安装一下夜莺并进行对接。当然，这里也可以对接自己内部的监控系统，不过需要实现一些接口。
 
 具体的文档可见：[监控功能对接夜莺](../dev_guide/monitor_system_integrate_with_n9e.md)、[监控功能对接其他系统](../dev_guide/monitor_system_integrate_with_self.md)
 
-### 11、连接信息、耗时信息为什么没有数据？
+那么在夜莺中，如何创建告警组呢？
+需要前往夜莺平台-用户资源中心-团队管理中新建团队。新建过团队之后再次回到Logi-KM中刷新页面就可以在该下拉框中选择告警接收组了。
+
+### 11、连接信息、耗时信息、磁盘信息为什么没有数据？
 
 这块需要结合滴滴内部的kafka-gateway一同使用才会有数据，滴滴kafka-gateway暂未开源。
 
@@ -126,3 +135,68 @@
 逻辑集群的申请与审批仅仅只是一个工单流程，并不会去实际创建逻辑集群，逻辑集群的创建还需要手动去创建。
 
 具体的操作可见：[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)

docs/user_guide/monitor_desc.md

@@ -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) |  |

docs/user_guide/user_guide_cn.md

@@ -609,10 +609,10 @@ Lag：表示该消费客户端是否有堆积；等于 partition offset-consume
 
 集群类型：选择创建的集群为“独享”还是“独立”。
 
-* 独享集群意味着，您独自拥有一个
+* 独享集群意味着， 您拥有一个集群中，部分broker的使用权限。
-集群；
 
-* 独立集群意味着，您拥有一个集群中，部分broker的使用权限。
+* 独立集群意味着，您独自拥有一个
+  物理集群；
 
 * <font color = red>共享集群意味着，大家共用一个集群及其中broker。</font>
 

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

@@ -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;
-    }
-}

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

@@ -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{" +

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

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

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

@@ -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{" +

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

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

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

@@ -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{" +

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

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

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

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

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

@@ -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{" +

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

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

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

@@ -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 code >= FINISHED.getCode();
-            return true;
-        }
-        return false;
     }
 }

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

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

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

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

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

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

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

@@ -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{" +

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

@@ -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_CLUSTER_ID("SD_CLUSTER_ID", "SD_CLUSTER_ID"),
-    SD_QUEUE_SIZE("SERVICE_DISCOVERY_QUEUE_SIZE", "SERVICE_DISCOVERY_QUEUE_SIZE"),
+    SD_QUEUE_SIZE("SD_QUEUE_SIZE", "SD_QUEUE_SIZE"),
-    SD_APP_ID_RATE("SERVICE_DISCOVERY_APPID_RATE", "SERVICE_DISCOVERY_APPID_RATE"),
+    SD_APP_RATE("SD_APP_RATE", "SD_APP_RATE"),
-    SD_IP_RATE("SERVICE_DISCOVERY_IP_RATE", "SERVICE_DISCOVERY_IP_RATE"),
+    SD_IP_RATE("SD_IP_RATE", "SD_IP_RATE"),
-    SD_SP_RATE("SERVICE_DISCOVERY_SP_RATE", "SERVICE_DISCOVERY_SP_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{" +

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

@@ -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() {
+    }
 }

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

@@ -20,6 +20,15 @@ public class ApiPrefix {
     // open
     public static final String API_V1_THIRD_PART_PREFIX = API_V1_PREFIX + "third-part/";
 
+    // 开放给OP的接口, 后续对 应的接口的集群都需要是物理集群
+    public static final String API_V1_THIRD_PART_OP_PREFIX = API_V1_THIRD_PART_PREFIX + "op/";
+
+    // 开放给Normal的接口, 后续对应的接口的集群，都需要是逻辑集群
+    public static final String API_V1_THIRD_PART_NORMAL_PREFIX = API_V1_THIRD_PART_PREFIX + "normal/";
+
     // gateway
     public static final String GATEWAY_API_V1_PREFIX = "/gateway" + API_V1_PREFIX;
+
+    private ApiPrefix() {
+    }
 }

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

@@ -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() {
+    }
 }

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

@@ -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() {
+    }
 }

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

@@ -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() {
+    }
 }

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

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

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

@@ -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() {
+    }
 }

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

@@ -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() {
+    }
 }

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

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

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

@@ -54,4 +54,7 @@ public class TopicCreationConstant {
      * 单次自动化审批, 最多允许的通过单子
      */
     public static final Integer MAX_PASSED_ORDER_NUM_PER_TASK = 200;
+
+    private TopicCreationConstant() {
+    }
 }

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

@@ -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() {
+    }
 }

kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/constant/TrickLoginConstant.java

@@ -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() {
+    }
+}

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

@@ -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;
+    }
+
 }

kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/entity/dto/gateway/TopicQuotaDTO.java

@@ -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);
+    }
+}

kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/entity/vo/common/BrokerOverviewVO.java

@@ -33,7 +33,7 @@ public class BrokerOverviewVO {
     @ApiModelProperty(value = "分区数")
     private Integer partitionCount;
 
-    @ApiModelProperty(value = "已同步副本数")
+    @ApiModelProperty(value = "失效副本分区的个数")
     private Integer underReplicatedPartitions;
 
     @ApiModelProperty(value = "未同步")

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

@@ -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 =
+        HttpServletRequest request = ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getRequest();
-                ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getRequest();
+
-        HttpSession session = request.getSession();
+        String username = null;
-        String username = (String) session.getAttribute(LoginConstant.SESSION_USERNAME_KEY);
+        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;
         }

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

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

kafka-manager-console/package.json

@@ -1,6 +1,6 @@
 {
   "name": "logi-kafka",
-  "version": "2.3.1",
+  "version": "v2.4.2",
   "description": "",
   "scripts": {
     "start": "webpack-dev-server",

kafka-manager-console/src/container/admin/config.tsx

@@ -28,14 +28,16 @@ export const getUserColumns = () => {
           <span className="table-operation">
             <a onClick={() => showApplyModal(record)}>编辑</a>
             <a onClick={() => showApplyModalModifyPassword(record)}>修改密码</a>
-            <Popconfirm
+            {record.username == users.currentUser.username ? "" :
-              title="确定删除？"
+                <Popconfirm
-              onConfirm={() => users.deleteUser(record.username)}
+                  title="确定删除？"
-              cancelText="取消"
+                  onConfirm={() => users.deleteUser(record.username)}
-              okText="确认"
+                  cancelText="取消"
-            >
+                  okText="确认"
-              <a>删除</a>
+                >
-            </Popconfirm>
+                  <a>删除</a>
+                </Popconfirm>
+            }
           </span>);
       },
     },

kafka-manager-console/src/container/header/index.tsx

@@ -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">

kafka-manager-console/src/container/modal/order.tsx

@@ -10,8 +10,8 @@ import * as React from 'react';
 
 const updateInputModal = (status: string, type: number) => {
   const formMap = wrapper.xFormWrapper.formMap;
-  const region = type === 0 ? 5 : 3;
+  const region = type === 0 ? 6 : 3;
-  const broker = type === 0 ? 6 : 4;
+  const broker = type === 0 ? 7 : 4;
   formMap[region].invisible = status === 'region';
   formMap[broker].invisible = status !== 'region';
 

kafka-manager-console/src/container/search-filter.tsx

@@ -126,7 +126,7 @@ export class SearchAndFilterContainer extends React.Component<any, ISearchAndFil
     );
   }
 
-  public renderSearch(text?: string, placeholder?: string, keyName: string = 'searchKey',) {
+  public renderSearch(text?: string, placeholder?: string, keyName: string = 'searchKey') {
     const value = this.state[keyName] as string;
     return (
       <li className="render-box">

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

@@ -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);
 }
 

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

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

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

@@ -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);
 }

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

@@ -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_RATE.getConfigType(),
-                    GatewayConfigKeyEnum.SD_APP_ID_RATE.getConfigName()
+                    GatewayConfigKeyEnum.SD_APP_RATE.getConfigName()
             );
             if (ValidateUtils.isNull(configDO) || configDO.getVersion() <= requestVersion) {
                 return new AppRateConfig(Long.MIN_VALUE, null);

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

@@ -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;
+    }
 }

kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/service/gateway/impl/TopicConnectionServiceImpl.java

@@ -26,21 +26,27 @@ public class TopicConnectionServiceImpl implements TopicConnectionService {
     @Autowired
     private TopicConnectionDao topicConnectionDao;
 
+    private int splitInterval = 50;
+
     @Override
     public void batchAdd(List<TopicConnectionDO> doList) {
         if (ValidateUtils.isEmptyList(doList)) {
             return;
         }
+        int allSize = doList.size();
+        int successSize = 0;
 
-        int count = 0;
+        int part = doList.size() / splitInterval;
-        for (TopicConnectionDO connectionDO: doList) {
+        for (int i = 0; i < part; ++i) {
-            try {
+            List<TopicConnectionDO> subList = doList.subList(0, splitInterval);
-                count += topicConnectionDao.replace(connectionDO);
+            successSize += topicConnectionDao.batchReplace(subList);
-            } catch (Exception e) {
+            doList.subList(0, splitInterval).clear();
-                LOGGER.error("class=TopicConnectionServiceImpl||method=batchAdd||connectionDO={}||errMsg={}", connectionDO, e.getMessage());
-            }
         }
-        LOGGER.info("class=TopicConnectionServiceImpl||method=batchAdd||allSize={}||successSize={}", doList.size(), count);
+        if (!ValidateUtils.isEmptyList(doList)) {
+            successSize += topicConnectionDao.batchReplace(doList);
+        }
+        LOGGER.info("class=TopicConnectionServiceImpl||method=batchAdd||allSize={}||successSize={}", allSize, successSize);
+
     }
 
     @Override

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

@@ -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
             );
         }

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

@@ -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,

kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/utils/ConfigUtils.java

@@ -17,7 +17,7 @@ public class ConfigUtils {
     private String kafkaManagerEnv;
 
     @Value(value = "${custom.store-metrics-task.save-days}")
-    private Integer maxMetricsSaveDays;
+    private Long maxMetricsSaveDays;
 
     public String getIdc() {
         return idc;
@@ -35,11 +35,11 @@ public class ConfigUtils {
         this.kafkaManagerEnv = kafkaManagerEnv;
     }
 
-    public Integer getMaxMetricsSaveDays() {
+    public Long getMaxMetricsSaveDays() {
         return maxMetricsSaveDays;
     }
 
-    public void setMaxMetricsSaveDays(Integer maxMetricsSaveDays) {
+    public void setMaxMetricsSaveDays(Long maxMetricsSaveDays) {
         this.maxMetricsSaveDays = maxMetricsSaveDays;
     }
 }

kafka-manager-dao/src/main/java/com/xiaojukeji/kafka/manager/dao/gateway/TopicConnectionDao.java

@@ -13,8 +13,6 @@ import java.util.List;
 public interface TopicConnectionDao {
     int batchReplace(List<TopicConnectionDO> doList);
 
-    int replace(TopicConnectionDO topicConnectionDO);
-
     List<TopicConnectionDO> getByTopicName(Long clusterId, String topicName, Date startTime, Date endTime);
 
     List<TopicConnectionDO> getByAppId(String appId, Date startTime, Date endTime);

kafka-manager-dao/src/main/java/com/xiaojukeji/kafka/manager/dao/gateway/impl/TopicConnectionDaoImpl.java

@@ -27,29 +27,12 @@ public class TopicConnectionDaoImpl implements TopicConnectionDao {
 
     @Override
     public int batchReplace(List<TopicConnectionDO> doList) {
-        int count = 0;
-        for (TopicConnectionDO elem: doList) {
-            try {
-                count += sqlSession.insert("TopicConnectionDao.replace", elem);
-            } catch (DeadlockLoserDataAccessException e1) {
-
-            } catch (Exception e) {
-                LOGGER.error("add topic connection info, clusterId:{} topicName:{}."
-                        , elem.getClusterId(), elem.getTopicName(), e);
-            }
-        }
-        return count;
-    }
-
-    @Override
-    public int replace(TopicConnectionDO topicConnectionDO) {
         try {
-            return sqlSession.insert("TopicConnectionDao.replace", topicConnectionDO);
+            return sqlSession.insert("TopicConnectionDao.batchReplace", doList);
         } catch (DeadlockLoserDataAccessException e1) {
             return 0;
         } catch (Exception e) {
-            LOGGER.error("add topic connection info, clusterId:{} topicName:{}."
+            LOGGER.error("add topic connections info failed", e);
-                    , topicConnectionDO.getClusterId(), topicConnectionDO.getTopicName(), e);
         }
         return 0;
     }

kafka-manager-dao/src/main/resources/mapper/TopicConnectionDao.xml

@@ -13,7 +13,7 @@
         <result property="createTime"       column="create_time"/>
     </resultMap>
 
-    <insert id="replace" parameterType="com.xiaojukeji.kafka.manager.common.entity.pojo.gateway.TopicConnectionDO">
+    <insert id="batchReplace" parameterType="java.util.List">
         REPLACE INTO topic_connections (
             cluster_id,
             topic_name,
@@ -22,16 +22,19 @@
             ip,
             client_version,
             create_time
-          )
+        )
-        VALUES (
+        VALUES
-            #{clusterId},
+        <foreach collection="list" item="item" index="index" separator=",">
-            #{topicName},
+            (
-            #{type},
+            #{item.clusterId},
-            #{appId},
+            #{item.topicName},
-            #{ip},
+            #{item.type},
-            #{clientVersion},
+            #{item.appId},
-            #{createTime}
+            #{item.ip},
-          )
+            #{item.clientVersion},
+            #{item.createTime}
+            )
+        </foreach>
     </insert>
 
     <select id="getByTopicName" parameterType="java.util.Map" resultMap="TopicConnectionMap">

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

@@ -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);
 }

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

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

kafka-manager-extends/kafka-manager-account/src/main/java/com/xiaojukeji/kafka/manager/account/component/login/trick/TrickLoginService.java

@@ -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();
+    }
+}

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

@@ -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;
 
@@ -67,20 +71,24 @@ public class LoginServiceImpl implements LoginService {
         if (ValidateUtils.isNull(classRequestMappingValue)) {
             LOGGER.error("class=LoginServiceImpl||method=checkLogin||msg=uri illegal||uri={}", request.getRequestURI());
             singleSignOn.setRedirectToLoginPage(response);
+            response.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
             return false;
         }
 
         if (classRequestMappingValue.equals(ApiPrefix.API_V1_SSO_PREFIX)
                 || classRequestMappingValue.equals(ApiPrefix.API_V1_THIRD_PART_PREFIX)
+                || classRequestMappingValue.equals(ApiPrefix.API_V1_THIRD_PART_OP_PREFIX)
+                || classRequestMappingValue.equals(ApiPrefix.API_V1_THIRD_PART_NORMAL_PREFIX)
                 || classRequestMappingValue.equals(ApiPrefix.GATEWAY_API_V1_PREFIX)) {
             // 白名单接口直接true
             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);
+            response.setStatus(HttpServletResponse.SC_UNAUTHORIZED);
             return false;
         }
 
@@ -111,4 +119,4 @@ public class LoginServiceImpl implements LoginService {
         }
         return true;
     }
-}
+}

kafka-manager-extends/kafka-manager-openapi/src/main/java/com/xiaojukeji/kafka/manager/openapi/common/dto/TopicAuthorityDTO.java

@@ -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);
+    }
+}

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

@@ -148,10 +148,12 @@ public abstract class AbstractScheduledTask<E extends Comparable> implements Sch
         }
         Collections.sort(allTaskList);
 
-        List<HeartbeatDO> hostList = heartbeatDao.selectActiveHosts(
+        List<HeartbeatDO> hostList = heartbeatDao.selectActiveHosts(new Date(now - ScheduledTaskConstant.HEARTBEAT_TIME));
-                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();
         }
 

kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/dispatch/op/AutoHandleTopicOrder.java

@@ -1,6 +1,5 @@
 package com.xiaojukeji.kafka.manager.task.dispatch.op;
 
-import com.alibaba.fastjson.JSON;
 import com.xiaojukeji.kafka.manager.bpm.OrderService;
 import com.xiaojukeji.kafka.manager.bpm.common.OrderStatusEnum;
 import com.xiaojukeji.kafka.manager.bpm.common.OrderTypeEnum;
@@ -11,6 +10,7 @@ import com.xiaojukeji.kafka.manager.common.constant.TopicCreationConstant;
 import com.xiaojukeji.kafka.manager.common.entity.ResultStatus;
 import com.xiaojukeji.kafka.manager.common.entity.ao.config.CreateTopicElemConfig;
 import com.xiaojukeji.kafka.manager.bpm.common.entry.apply.OrderExtensionApplyTopicDTO;
+import com.xiaojukeji.kafka.manager.common.utils.JsonUtils;
 import com.xiaojukeji.kafka.manager.common.utils.ListUtils;
 import com.xiaojukeji.kafka.manager.common.utils.ValidateUtils;
 import com.xiaojukeji.kafka.manager.common.entity.pojo.*;
@@ -23,6 +23,7 @@ import com.xiaojukeji.kafka.manager.task.component.EmptyEntry;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.stereotype.Component;
 
 import java.util.Arrays;
@@ -35,6 +36,7 @@ import java.util.Properties;
  */
 @Component
 @CustomScheduled(name = "autoHandleTopicOrder", cron = "0 0/1 * * * ?", threadNum = 1)
+@ConditionalOnProperty(prefix = "task.op.order-auto-exec", name = "topic-enabled", havingValue = "true", matchIfMissing = false)
 public class AutoHandleTopicOrder extends AbstractScheduledTask<EmptyEntry> {
     private static final Logger LOGGER = LoggerFactory.getLogger(LogConstant.SCHEDULED_TASK_LOGGER);
 
@@ -72,11 +74,8 @@ public class AutoHandleTopicOrder extends AbstractScheduledTask<EmptyEntry> {
 
         Integer maxPassedOrderNumPerTask = configService.getAutoPassedTopicApplyOrderNumPerTask();
         for (OrderDO orderDO: doList) {
-            if (!OrderTypeEnum.APPLY_TOPIC.getCode().equals(orderDO.getType())) {
-                continue;
-            }
             try {
-                if (!handleApplyTopicOrder(orderDO)) {
+                if (!OrderTypeEnum.APPLY_TOPIC.getCode().equals(orderDO.getType()) && !handleApplyTopicOrder(orderDO)) {
                     continue;
                 }
                 maxPassedOrderNumPerTask -= 1;
@@ -91,7 +90,7 @@ public class AutoHandleTopicOrder extends AbstractScheduledTask<EmptyEntry> {
     }
 
     private boolean handleApplyTopicOrder(OrderDO orderDO) {
-        OrderExtensionApplyTopicDTO dto = JSON.parseObject(orderDO.getExtensions(), OrderExtensionApplyTopicDTO.class);
+        OrderExtensionApplyTopicDTO dto = JsonUtils.stringToObj(orderDO.getExtensions(), OrderExtensionApplyTopicDTO.class);
         Long physicalClusterId =
                 logicalClusterMetadataManager.getPhysicalClusterId(dto.getClusterId(), dto.isPhysicalClusterId());
 

kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/dispatch/op/AutomatedHandleOrder.java

@@ -20,6 +20,7 @@ import com.xiaojukeji.kafka.manager.task.component.EmptyEntry;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.stereotype.Component;
 
 import java.util.*;
@@ -31,6 +32,7 @@ import java.util.*;
  */
 @Component
 @CustomScheduled(name = "automatedHandleOrder", cron = "0 0/1 * * * ?", threadNum = 1)
+@ConditionalOnProperty(prefix = "task.op.order-auto-exec", name = "app-enabled", havingValue = "true", matchIfMissing = false)
 public class AutomatedHandleOrder extends AbstractScheduledTask<EmptyEntry> {
     private static final Logger LOGGER = LoggerFactory.getLogger(LogConstant.SCHEDULED_TASK_LOGGER);
 
@@ -51,6 +53,7 @@ public class AutomatedHandleOrder extends AbstractScheduledTask<EmptyEntry> {
     public void processTask(EmptyEntry entryEntry) {
         List<OrderDO> waitDealOrderList = orderService.getWaitDealOrder();
         if (ValidateUtils.isEmptyList(waitDealOrderList)) {
+            LOGGER.info("class=AutomatedHandleOrder||method=processTask||msg=waiting deal order is empty");
             return;
         }
 
@@ -65,17 +68,18 @@ public class AutomatedHandleOrder extends AbstractScheduledTask<EmptyEntry> {
     }
 
     private void handleAppApplyOrder(List<OrderDO> waitDealOrderList, List<OrderDO> passedOrderList) {
-        LOGGER.info("start handle app apply order.");
+        LOGGER.info("class=AutomatedHandleOrder||method=processTask||msg=start handle app apply order");
         if (ValidateUtils.isEmptyList(waitDealOrderList)) {
             return;
         }
+
         Integer maxNum = Constant.HANDLE_APP_APPLY_MAX_NUM_DEFAULT;
         ConfigDO configDO = configService.getByKey(Constant.HANDLE_APP_APPLY_MAX_NUM);
         if (!ValidateUtils.isNull(configDO)) {
             try {
                 maxNum = Integer.parseInt(configDO.getConfigValue());
             } catch (Exception e) {
-                LOGGER.error("", e);
+                LOGGER.error("class=AutomatedHandleOrder||method=processTask||configDO={}||msg=config value illegal", configDO, e);
             }
         }
         int handleNum = Math.min(maxNum - passedOrderList.size(), waitDealOrderList.size());
@@ -97,7 +101,7 @@ public class AutomatedHandleOrder extends AbstractScheduledTask<EmptyEntry> {
                     applyAppOrder.updateOrder(orderDO, baseDTO, Constant.AUTO_HANDLE_USER_NAME);
                 }
             } catch (Exception e) {
-                LOGGER.error("", e);
+                LOGGER.error("class=AutomatedHandleOrder||method=processTask||orderDO={}||msg=auto handle app order failed", orderDO, e);
             }
         }
     }

kafka-manager-web/pom.xml

@@ -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>

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

@@ -337,7 +337,7 @@ public class NormalTopicController {
         }
 
         return new Result<>(TopicModelConverter.convert2TopicMineAppVOList(
-                topicManagerService.getTopicMineApps(physicalClusterId, topicName, SpringTool.getUserName()))
+            topicManagerService.getTopicMineApps(physicalClusterId, topicName, SpringTool.getUserName()))
         );
     }
 

kafka-manager-web/src/main/java/com/xiaojukeji/kafka/manager/web/api/versionone/op/OpAuthorityController.java

@@ -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)));
+    }
+}

kafka-manager-web/src/main/java/com/xiaojukeji/kafka/manager/web/api/versionone/op/OpLeaderController.java

@@ -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_PREFIX)
+@RequestMapping(ApiPrefix.API_V1_OP_PREFIX)
-public class ThirdPartOpController {
+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()) {

kafka-manager-web/src/main/java/com/xiaojukeji/kafka/manager/web/api/versionone/op/OpQuotaController.java

@@ -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)));
+    }
+}

kafka-manager-web/src/main/java/com/xiaojukeji/kafka/manager/web/api/versionone/op/OpTopicController.java

@@ -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.constant.ApiPrefix;
-import com.xiaojukeji.kafka.manager.common.bizenum.TaskStatusEnum;
 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.TopicCreationDTO;
-import com.xiaojukeji.kafka.manager.common.entity.dto.op.topic.*;
+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 {
+public class OpTopicController {
-    @Autowired
-    private ClusterService clusterService;
-
     @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);
     }
 }
-

kafka-manager-web/src/main/java/com/xiaojukeji/kafka/manager/web/api/versionone/thirdpart/ThirdPartBrokerController.java

@@ -32,7 +32,7 @@ import java.util.stream.Collectors;
  */
 @Api(tags = "开放接口-Broker相关接口(REST)")
 @RestController
-@RequestMapping(ApiPrefix.API_V1_THIRD_PART_PREFIX)
+@RequestMapping(ApiPrefix.API_V1_THIRD_PART_OP_PREFIX)
 public class ThirdPartBrokerController {
     @Autowired
     private BrokerService brokerService;

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

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

kafka-manager-web/src/main/java/com/xiaojukeji/kafka/manager/web/converters/AuthorityConverter.java

@@ -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;
+    }
+}

kafka-manager-web/src/main/resources/application.yml

@@ -11,9 +11,9 @@ spring:
     name: kafkamanager
   datasource:
     kafka-manager:
-      jdbc-url: jdbc:mysql://127.0.0.1:3306/logi_kafka_manager?characterEncoding=UTF-8&useSSL=false&serverTimezone=GMT%2B8
+      jdbc-url: jdbc:mysql://localhost:3306/logi_kafka_manager?characterEncoding=UTF-8&useSSL=false&serverTimezone=GMT%2B8
-      username: admin
+      username: root
-      password: admin
+      password: 123456
       driver-class-name: com.mysql.cj.jdbc.Driver
   main:
     allow-bean-definition-overriding: true
@@ -46,6 +46,9 @@ custom:
 task:
   op:
     sync-topic-enabled: false # 未落盘的Topic定期同步到DB中
+    order-auto-exec:          # 工单自动化审批线程的开关
+      topic-enabled: false      # Topic工单自动化审批开关, false:关闭自动化审批, true:开启
+      app-enabled: false        # App工单自动化审批开关, false:关闭自动化审批, true:开启
 
 account:
   ldap:

pom.xml

@@ -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>
@@ -203,7 +203,7 @@
             <dependency>
                 <groupId>org.eclipse.jetty</groupId>
                 <artifactId>jetty-util</artifactId>
-                <version>9.2.24.v20180105</version>
+                <version>9.4.39.v20210325</version>
             </dependency>
 
             <dependency>
@@ -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>