开启集群数据权限，同一浏览器不同账号登录看到其它账号集群信息bug fixed.

send message with header

README.md

@@ -1,7 +1,7 @@
 # kafka可视化管理平台
 一款轻量级的kafka可视化管理平台，安装配置快捷、简单易用。  
 为了开发的省事，没有国际化支持，页面只支持中文展示。  
-用过rocketmq-console吧，对，前端展示风格跟那个有点类似。
+用过rocketmq-console(rocketmq-dashboard)吧，对，前端展示风格跟那个有点类似。
 
 ## 页面预览
 如果github能查看图片的话，可以点击[查看菜单页面](./document/overview/概览.md)，查看每个页面的样子
@@ -25,11 +25,11 @@ v1.0.6版本之前，如果kafka集群启用了ACL，但是控制台没看到Acl
 ![功能特性](./document/img/功能特性.png)
 
 ## 安装包下载
-点击下载(v1.0.7版本)：[kafka-console-ui.zip](https://github.com/xxd763795151/kafka-console-ui/releases/download/v1.0.7/kafka-console-ui.zip)  
+点击下载(v1.0.9版本)：[kafka-console-ui.zip](https://github.com/xxd763795151/kafka-console-ui/releases/download/v1.0.9/kafka-console-ui-1.0.9.zip)  
 
 如果安装包下载的比较慢，可以查看下面的源码打包说明，把代码下载下来，本地快速打包.  
 
-github下载慢也可以试试从gitee下载，点击下载[gitee来源kafka-console-ui.zip](https://gitee.com/xiaodong_xu/kafka-console-ui/releases/download/v1.0.7/kafka-console-ui.zip)
+github下载慢也可以试试从gitee下载，点击下载[gitee来源kafka-console-ui.zip](https://gitee.com/xiaodong_xu/kafka-console-ui/releases/download/v1.0.9/kafka-console-ui-1.0.9.zip)
 
 ## 快速使用
 ### Windows
@@ -79,14 +79,29 @@ sh bin/shutdown.sh
 如果需要本地开发，开发环境配置查看：[本地开发](./document/develop/开发配置.md)
 
 ## 登录认证和权限
-目前主分支不支持登录认证，感谢@dongyinuo 同学开发了一版支持登录认证，及相关的按钮权限（主要有两个角色：管理员和普通开发人员）。  
+1.0.7版本及之前，主分支不支持登录认证，感谢@dongyinuo 同学开发了一版支持登录认证，及相关的按钮权限（主要有两个角色：管理员和普通开发人员）。  
 在分支：feature/dongyinuo/20220501/devops 上。  
 如果有需要使用管理台登录认证的，可以切换到这个分支上进行打包，打包方式看 源码打包 说明。  
 默认登录账户：admin/kafka-console-ui521
 
+目前最新版本主分支已增加登录认证、用户、角色等权限配置，如需开启登录认证，修改配置文件config/application.yml: auth.enable=true(默认 false)，如下：
+```yaml
+# 权限认证设置，设置为true，需要先登录才能访问
+auth:
+  enable: true
+  # 登录用户token的过期时间，单位：小时
+  expire-hours: 24
+```
+默认有两个登录用户：super-admin/123456，admin/123456，登录成功后在个人设置修改密码。super-admin和admin的唯一区别是super-admin可以增加删除用户，admin不能。如果觉得不合适，请在用户菜单下删除相关用户或角色自行创建合适的角色或用户。
+注意：不开启登录认证，页面不显示用户菜单，正常现象。
+
+
 ## DockerCompose部署
 感谢@wdkang123 同学分享的部署方式，如果有需要请查看[DockerCompose部署方式](./document/deploy/docker部署.md)
+## 感谢支持
+感谢jetbrains的开源支持，如果有朋友愿意一起维护，很欢迎提pr.
 
+[![jetbrains](./document/img/jb_beam.svg "jetbrains")](https://jb.gg/OpenSourceSupport)
 ## 联系方式
 + 微信群
 

assembly.xml

@@ -2,10 +2,10 @@
     xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.2"
     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
     xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.2 http://maven.apache.org/xsd/assembly-1.1.2.xsd">
-    <id>rocketmq-reput</id>
+    <id>${project.version}</id>
     <formats>
-<!--        <format>tar.gz</format>-->
         <format>zip</format>
+        <format>tar.gz</format>
     </formats>
 
     <fileSets>

bin/shutdown.sh

@@ -1,7 +1,12 @@
 #!/bin/bash
 
-SCRIPT_DIR=`dirname $0`
+PREFIX='./'
-PROJECT_DIR="$SCRIPT_DIR/.."
+CMD=$0
+if [[ $CMD == $PREFIX* ]]; then
+        CMD=${CMD:2}
+fi
+SCRIPT_DIR=$(dirname "`pwd`/$CMD")
+PROJECT_DIR=`dirname "$SCRIPT_DIR"`
 # 不要修改进程标记，作为进程属性关闭使用
 PROCESS_FLAG="kafka-console-ui-process-flag:${PROJECT_DIR}"
 pkill -f $PROCESS_FLAG

bin/start.bat

@@ -1,7 +1,7 @@
 rem MAIN_CLASS=org.springframework.boot.loader.JarLauncher
 rem JAVA_HOME=jre1.8.0_66
 set JAVA_CMD=%JAVA_HOME%\bin\java
-set JAVA_OPTS=-Xmx512m -Xms512m -Xmn256m -Xss256k
+set JAVA_OPTS=-Xmx512m -Xms512m -Xmn256m -Xss256k -Dfile.encoding=utf-8
 set CONFIG_FILE=../config/application.yml
 set TARGET=../lib/kafka-console-ui.jar
 set DATA_DIR=..

bin/start.sh

@@ -3,8 +3,13 @@
 # 设置jvm堆大小及栈大小，栈大小最少设置为256K，不要小于这个值，比如设置为128，太小了
 JAVA_MEM_OPTS="-Xmx512m -Xms512m -Xmn256m -Xss256k"
 
-SCRIPT_DIR=`dirname $0`
+PREFIX='./'
-PROJECT_DIR="$SCRIPT_DIR/.."
+CMD=$0
+if [[ $CMD == $PREFIX* ]]; then
+        CMD=${CMD:2}
+fi
+SCRIPT_DIR=$(dirname "`pwd`/$CMD")
+PROJECT_DIR=`dirname "$SCRIPT_DIR"`
 CONF_FILE="$PROJECT_DIR/config/application.yml"
 TARGET="$PROJECT_DIR/lib/kafka-console-ui.jar"
 
@@ -12,12 +17,12 @@ TARGET="$PROJECT_DIR/lib/kafka-console-ui.jar"
 DATA_DIR=$PROJECT_DIR
 
 # 日志目录，默认为当前工程目录下
-# 这个是错误输出，如果启动命令有误，输出到这个文件，应用日志不会输出到error.out，应用日志输出到上面的rocketmq-reput.log中
+# 这个是错误输出，如果启动命令有误，输出到这个文件，应用日志不会输出到error.out，应用日志输出到上面的kafka-console-ui.log中
 ERROR_OUT="$PROJECT_DIR/error.out"
-# 不要修改进程标记，作为进程属性关闭使用，如果要修改，请把stop.sh里的该属性的值保持一致
+# 不要修改进程标记，作为进程属性关闭使用，如果要修改，请把shutdown.sh里的该属性的值保持一致
 PROCESS_FLAG="kafka-console-ui-process-flag:${PROJECT_DIR}"
 
-JAVA_OPTS="$JAVA_OPTS $JAVA_MEM_OPTS"
+JAVA_OPTS="$JAVA_OPTS $JAVA_MEM_OPTS -Dfile.encoding=utf-8"
 
 nohup java -jar $JAVA_OPTS $TARGET --spring.config.location="$CONF_FILE" --logging.home="$PROJECT_DIR" --data.dir=$DATA_DIR $PROCESS_FLAG 1>/dev/null 2>$ERROR_OUT &
 

document/img/jb_beam.svg

@@ -0,0 +1 @@
+<svg height="180" viewBox="0 0 180 180" width="180" xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink"><linearGradient id="a" gradientUnits="userSpaceOnUse" x1="32.64" x2="82.77" y1="61.16" y2="85.54"><stop offset=".21" stop-color="#fe2857"/><stop offset="1" stop-color="#293896"/></linearGradient><linearGradient id="b" gradientUnits="userSpaceOnUse" x1="17.38" x2="82.95" y1="69.86" y2="21.23"><stop offset="0" stop-color="#fe2857"/><stop offset=".01" stop-color="#fe2857"/><stop offset=".86" stop-color="#ff318c"/></linearGradient><linearGradient id="c" gradientUnits="userSpaceOnUse" x1="74.17" x2="160.27" y1="21.58" y2="99.76"><stop offset=".02" stop-color="#ff318c"/><stop offset=".21" stop-color="#fe2857"/><stop offset=".86" stop-color="#fdb60d"/></linearGradient><linearGradient id="d" gradientUnits="userSpaceOnUse" x1="155.46" x2="55.07" y1="89.8" y2="158.9"><stop offset=".01" stop-color="#fdb60d"/><stop offset=".86" stop-color="#fcf84a"/></linearGradient><path d="m81.56 83.71-41.35-35a15 15 0 1 0 -14.47 25.7h.15l.39.12 52.16 15.89a3.53 3.53 0 0 0 1.18.21 3.73 3.73 0 0 0 1.93-6.91z" fill="url(#a)"/><path d="m89.85 25.93a10.89 10.89 0 0 0 -16.85-9.18l-50.5 30.66a15 15 0 1 0 17.9 24l45.27-36.89.36-.3a10.93 10.93 0 0 0 3.82-8.29z" fill="url(#b)"/><path d="m163.29 92-76.62-73.79a10.91 10.91 0 1 0 -14.81 16l.14.12 81.4 68.58a7.36 7.36 0 0 0 12.09-5.65 7.39 7.39 0 0 0 -2.2-5.26z" fill="url(#c)"/><path d="m165.5 97.29a7.35 7.35 0 0 0 -11.67-6l-92.71 45.3a15 15 0 1 0 15.48 25.59l85.73-58.84a7.35 7.35 0 0 0 3.17-6.05z" fill="url(#d)"/><path d="m60 60h60v60h-60z"/><g fill="#fff"><path d="m66.53 108.75h22.5v3.75h-22.5z"/><path d="m65.59 75.47 1.67-1.58a1.88 1.88 0 0 0 1.47.87c.64 0 1.06-.45 1.06-1.32v-5.92h2.58v5.94a3.44 3.44 0 0 1 -.92 2.63 3.52 3.52 0 0 1 -2.57 1 3.84 3.84 0 0 1 -3.29-1.62z"/><path d="m73.53 67.52h7.53v2.19h-5v1.43h4.49v2h-4.45v1.49h5v2.2h-7.6z"/><path d="m84.73 69.79h-2.8v-2.27h8.21v2.27h-2.81v7.09h-2.6z"/><path d="m66.63 80.58h4.42a3.47 3.47 0 0 1 2.55.83 2.09 2.09 0 0 1 .61 1.52 2.18 2.18 0 0 1 -1.45 2.09 2.27 2.27 0 0 1 1.86 2.29c0 1.69-1.31 2.69-3.55 2.69h-4.44zm5 2.89c0-.52-.42-.8-1.18-.8h-1.29v1.64h1.25c.78 0 1.24-.27 1.24-.81zm-.9 2.66h-1.57v1.73h1.62c.8 0 1.24-.31 1.24-.86-.02-.53-.4-.87-1.27-.87z"/><path d="m75.45 80.58h4.15a4.14 4.14 0 0 1 3.05 1 2.92 2.92 0 0 1 .83 2.18 3 3 0 0 1 -1.93 2.89l2.24 3.35h-3l-1.89-2.84h-.87v2.84h-2.6zm4 4.5c.87 0 1.4-.43 1.4-1.12 0-.75-.55-1.13-1.41-1.13h-1.39v2.27z"/><path d="m87.09 80.51h2.5l4 9.44h-2.79l-.67-1.69h-3.63l-.67 1.74h-2.71zm2.28 5.73-1.05-2.65-1.06 2.65z"/><path d="m94 80.55h2.6v9.37h-2.6z"/><path d="m97.56 80.55h2.44l3.37 5v-5h2.57v9.37h-2.27l-3.53-5.14v5.14h-2.58z"/><path d="m106.37 88.53 1.44-1.73a4.86 4.86 0 0 0 3 1.13c.71 0 1.08-.25 1.08-.65 0-.41-.3-.61-1.59-.91-2-.46-3.53-1-3.53-2.93 0-1.74 1.38-3 3.63-3a5.88 5.88 0 0 1 3.85 1.25l-1.25 1.78a4.56 4.56 0 0 0 -2.62-.92c-.63 0-.94.25-.94.6 0 .43.32.62 1.63.91 2.15.47 3.48 1.17 3.48 2.92 0 1.91-1.51 3-3.78 3a6.56 6.56 0 0 1 -4.4-1.45z"/></g><path d="m0 0h180v180h-180z" fill="none"/></svg>

pom.xml

@@ -10,7 +10,7 @@
     </parent>
     <groupId>com.xuxd</groupId>
     <artifactId>kafka-console-ui</artifactId>
-    <version>1.0.7</version>
+    <version>1.1.0</version>
     <name>kafka-console-ui</name>
     <description>Kafka console manage ui</description>
     <properties>
@@ -21,7 +21,7 @@
         <ui.path>${project.basedir}/ui</ui.path>
         <frontend-maven-plugin.version>1.11.0</frontend-maven-plugin.version>
         <compiler.version>1.8</compiler.version>
-        <kafka.version>3.2.0</kafka.version>
+        <kafka.version>3.5.0</kafka.version>
         <maven.assembly.plugin.version>3.0.0</maven.assembly.plugin.version>
         <mybatis-plus-boot-starter.version>3.4.2</mybatis-plus-boot-starter.version>
         <scala.version>2.13.6</scala.version>
@@ -90,6 +90,12 @@
             </exclusions>
         </dependency>
 
+<!--        <dependency>-->
+<!--            <groupId>org.apache.kafka</groupId>-->
+<!--            <artifactId>kafka-tools</artifactId>-->
+<!--            <version>${kafka.version}</version>-->
+<!--        </dependency>-->
+
         <dependency>
             <groupId>com.typesafe.scala-logging</groupId>
             <artifactId>scala-logging_2.13</artifactId>
@@ -174,26 +180,26 @@
                         </executions>
                     </plugin>
 
-<!--                    <plugin>-->
+                    <!--                    <plugin>-->
-<!--                        <groupId>org.codehaus.mojo</groupId>-->
+                    <!--                        <groupId>org.codehaus.mojo</groupId>-->
-<!--                        <artifactId>build-helper-maven-plugin</artifactId>-->
+                    <!--                        <artifactId>build-helper-maven-plugin</artifactId>-->
-<!--                        <version>3.2.0</version>-->
+                    <!--                        <version>3.2.0</version>-->
-<!--                        <executions>-->
+                    <!--                        <executions>-->
-<!--                            <execution>-->
+                    <!--                            <execution>-->
-<!--                                <id>add-source</id>-->
+                    <!--                                <id>add-source</id>-->
-<!--                                <phase>generate-sources</phase>-->
+                    <!--                                <phase>generate-sources</phase>-->
-<!--                                <goals>-->
+                    <!--                                <goals>-->
-<!--                                    <goal>add-source</goal>-->
+                    <!--                                    <goal>add-source</goal>-->
-<!--                                </goals>-->
+                    <!--                                </goals>-->
-<!--                                <configuration>-->
+                    <!--                                <configuration>-->
-<!--                                    <sources>-->
+                    <!--                                    <sources>-->
-<!--                                        <source>src/main/java</source>-->
+                    <!--                                        <source>src/main/java</source>-->
-<!--                                        <source>src/main/scala</source>-->
+                    <!--                                        <source>src/main/scala</source>-->
-<!--                                    </sources>-->
+                    <!--                                    </sources>-->
-<!--                                </configuration>-->
+                    <!--                                </configuration>-->
-<!--                            </execution>-->
+                    <!--                            </execution>-->
-<!--                        </executions>-->
+                    <!--                        </executions>-->
-<!--                    </plugin>-->
+                    <!--                    </plugin>-->
 
                     <plugin>
                         <artifactId>maven-compiler-plugin</artifactId>
@@ -201,6 +207,8 @@
                             <source>${compiler.version}</source>
                             <target>${compiler.version}</target>
                             <encoding>${project.build.sourceEncoding}</encoding>
+<!--                            <debug>false</debug>-->
+<!--                            <parameters>false</parameters>-->
                         </configuration>
                     </plugin>
                     <plugin>
@@ -225,7 +233,7 @@
                                     <goal>npm</goal>
                                 </goals>
                                 <configuration>
-<!--                                    <arguments>install &#45;&#45;registry=https://registry.npmjs.org/</arguments>-->
+                                    <!--                                    <arguments>install &#45;&#45;registry=https://registry.npmjs.org/</arguments>-->
                                     <arguments>install --registry=https://registry.npm.taobao.org</arguments>
                                 </configuration>
                             </execution>
@@ -279,7 +287,7 @@
                             <attach>true</attach>
                             <tarLongFileMode>posix</tarLongFileMode>
                             <runOnlyAtExecutionRoot>false</runOnlyAtExecutionRoot>
-                            <appendAssemblyId>false</appendAssemblyId>
+                            <appendAssemblyId>true</appendAssemblyId>
                         </configuration>
                     </plugin>
                 </plugins>

src/main/java/com/xuxd/kafka/console/aspect/ControllerLogAspect.java

@@ -1,6 +1,9 @@
 package com.xuxd.kafka.console.aspect;
 
 import com.xuxd.kafka.console.aspect.annotation.ControllerLog;
+import com.xuxd.kafka.console.beans.Credentials;
+import com.xuxd.kafka.console.config.LogConfig;
+import com.xuxd.kafka.console.filter.CredentialsContext;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.aspectj.lang.ProceedingJoinPoint;
@@ -12,6 +15,7 @@ import org.springframework.stereotype.Component;
 
 import java.lang.reflect.Method;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.locks.ReentrantLock;
 
@@ -28,6 +32,12 @@ public class ControllerLogAspect {
 
     private ReentrantLock lock = new ReentrantLock();
 
+    private final LogConfig logConfig;
+
+    public ControllerLogAspect(LogConfig logConfig) {
+        this.logConfig = logConfig;
+    }
+
     @Pointcut("@annotation(com.xuxd.kafka.console.aspect.annotation.ControllerLog)")
     private void pointcut() {
 
@@ -35,6 +45,9 @@ public class ControllerLogAspect {
 
     @Around("pointcut()")
     public Object around(ProceedingJoinPoint joinPoint) throws Throwable {
+        if (!logConfig.isPrintControllerLog()) {
+            return joinPoint.proceed();
+        }
         StringBuilder params = new StringBuilder("[");
         try {
             String methodName = getMethodFullName(joinPoint.getTarget().getClass().getName(), joinPoint.getSignature().getName());
@@ -56,6 +69,10 @@ public class ControllerLogAspect {
             String resStr = "[" + (res != null ? res.toString() : "") + "]";
 
             StringBuilder sb = new StringBuilder();
+            Credentials credentials = CredentialsContext.get();
+            if (credentials != null) {
+                sb.append("[").append(credentials.getUsername()).append("] ");
+            }
             String shortMethodName = descMap.getOrDefault(methodName, ".-");
             shortMethodName = shortMethodName.substring(shortMethodName.lastIndexOf(".") + 1);
             sb.append("[").append(shortMethodName)
@@ -85,6 +102,9 @@ public class ControllerLogAspect {
                 Class<?>[] clzArr = new Class[args.length];
                 for (int i = 0; i < args.length; i++) {
                     clzArr[i] = args[i].getClass();
+                    if (List.class.isAssignableFrom(clzArr[i])) {
+                        clzArr[i] = List.class;
+                    }
                 }
                 method = aClass.getDeclaredMethod(methodName, clzArr);
 

src/main/java/com/xuxd/kafka/console/aspect/PermissionAspect.java

@@ -101,17 +101,30 @@ public class PermissionAspect {
             throw new UnAuthorizedException(credentials.getUsername() + ":" + allowPermSet);
         }
 
+        boolean unauthorized = true;
+        boolean notFoundHideProperty = true;
         String roleIds = userDO.getRoleIds();
-        List<Long> roleIdList = Arrays.stream(roleIds.split(",")).map(String::trim).filter(StringUtils::isNotEmpty).map(Long::valueOf).collect(Collectors.toList());
+        List<Long> roleIdList = Arrays.stream(roleIds.split(",")).
+                map(String::trim).filter(StringUtils::isNotEmpty).
+                map(Long::valueOf).collect(Collectors.toList());
         for (Long roleId : roleIdList) {
             Set<String> permSet = rolePermCache.getRolePermCache().getOrDefault(roleId, Collections.emptySet());
             for (String p : allowPermSet) {
                 if (permSet.contains(p)) {
-                    return;
+                    unauthorized = false;
                 }
             }
+            if (permSet.contains(authConfig.getHideClusterPropertyPerm())) {
+                notFoundHideProperty = false;
+            }
         }
-        throw new UnAuthorizedException(credentials.getUsername() + ":" + allowPermSet);
+        if (unauthorized) {
+            throw new UnAuthorizedException(credentials.getUsername() + ":" + allowPermSet);
+        }
+        if (authConfig.isHideClusterProperty() && notFoundHideProperty) {
+            credentials.setHideClusterProperty(true);
+        }
+        credentials.setRoleIdList(roleIdList);
     }
 
     private Map<String, Set<String>> checkPermMap(String methodName, String[] value) {

src/main/java/com/xuxd/kafka/console/beans/Credentials.java

@@ -2,6 +2,8 @@ package com.xuxd.kafka.console.beans;
 
 import lombok.Data;
 
+import java.util.List;
+
 /**
  * @author: xuxd
  * @date: 2023/5/14 19:37
@@ -15,6 +17,13 @@ public class Credentials {
 
     private long expiration;
 
+    /**
+     * 是否隐藏集群属性
+     */
+    private boolean hideClusterProperty;
+
+    private List<Long> roleIdList;
+
     public boolean isInvalid() {
         return this == INVALID;
     }

src/main/java/com/xuxd/kafka/console/beans/QueryMessage.java

@@ -33,4 +33,6 @@ public class QueryMessage {
     private String headerKey;
 
     private String headerValue;
+
+    private int filterNumber;
 }

src/main/java/com/xuxd/kafka/console/beans/SendMessage.java

@@ -1,5 +1,6 @@
 package com.xuxd.kafka.console.beans;
 
+import java.util.List;
 import lombok.Data;
 
 /**
@@ -22,4 +23,18 @@ public class SendMessage {
     private int num;
 
     private long offset;
+
+    private List<Header> headers;
+
+    /**
+     * true: sync send.
+     */
+    private boolean sync;
+
+    @Data
+    public static class Header{
+        private String headerKey;
+
+        private String headerValue;
+    }
 }

src/main/java/com/xuxd/kafka/console/beans/dos/ClusterRoleRelationDO.java

@@ -0,0 +1,24 @@
+package com.xuxd.kafka.console.beans.dos;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.Data;
+
+/**
+ * @author: xuxd
+ * @since: 2023/8/23 21:35
+ **/
+@Data
+@TableName("t_cluster_role_relation")
+public class ClusterRoleRelationDO {
+
+    @TableId(type = IdType.AUTO)
+    private Long id;
+
+    private Long roleId;
+
+    private Long clusterInfoId;
+
+    private String updateTime;
+}

src/main/java/com/xuxd/kafka/console/beans/dto/ClusterRoleRelationDTO.java

@@ -0,0 +1,28 @@
+package com.xuxd.kafka.console.beans.dto;
+
+import com.xuxd.kafka.console.beans.dos.ClusterRoleRelationDO;
+import lombok.Data;
+
+/**
+ * @author: xuxd
+ * @since: 2023/8/23 21:42
+ **/
+@Data
+public class ClusterRoleRelationDTO {
+
+    private Long id;
+
+    private Long roleId;
+
+    private Long clusterInfoId;
+
+    private String updateTime;
+
+    public ClusterRoleRelationDO toDO() {
+        ClusterRoleRelationDO aDo = new ClusterRoleRelationDO();
+        aDo.setId(id);
+        aDo.setRoleId(roleId);
+        aDo.setClusterInfoId(clusterInfoId);
+        return aDo;
+    }
+}

src/main/java/com/xuxd/kafka/console/beans/dto/QueryMessageDTO.java

@@ -37,6 +37,8 @@ public class QueryMessageDTO {
 
     private String headerValue;
 
+    private int filterNumber;
+
     public QueryMessage toQueryMessage() {
         QueryMessage queryMessage = new QueryMessage();
         queryMessage.setTopic(topic);
@@ -69,6 +71,7 @@ public class QueryMessageDTO {
         if (StringUtils.isNotBlank(headerValue)) {
             queryMessage.setHeaderValue(headerValue.trim());
         }
+        queryMessage.setFilterNumber(filterNumber);
 
         return queryMessage;
     }

src/main/java/com/xuxd/kafka/console/beans/dto/QuerySendStatisticsDTO.java

@@ -0,0 +1,23 @@
+package com.xuxd.kafka.console.beans.dto;
+
+import lombok.Data;
+
+import java.util.Date;
+import java.util.Set;
+
+/**
+ * 发送统计查询请求，指定时间段内，发送了多少消息.
+ * @author: xuxd
+ * @since: 2023/12/1 22:01
+ **/
+@Data
+public class QuerySendStatisticsDTO {
+
+    private String topic;
+
+    private Set<Integer> partition;
+
+    private Date startTime;
+
+    private Date endTime;
+}

src/main/java/com/xuxd/kafka/console/beans/vo/ClusterRoleRelationVO.java

@@ -0,0 +1,33 @@
+package com.xuxd.kafka.console.beans.vo;
+
+import com.xuxd.kafka.console.beans.dos.ClusterRoleRelationDO;
+import lombok.Data;
+
+/**
+ * @author: xuxd
+ * @since: 2023/8/23 21:45
+ **/
+@Data
+public class ClusterRoleRelationVO {
+
+    private Long id;
+
+    private Long roleId;
+
+    private Long clusterInfoId;
+
+    private String updateTime;
+
+    private String roleName;
+
+    private String clusterName;
+
+    public static ClusterRoleRelationVO from(ClusterRoleRelationDO relationDO) {
+        ClusterRoleRelationVO vo = new ClusterRoleRelationVO();
+        vo.setId(relationDO.getId());
+        vo.setRoleId(relationDO.getRoleId());
+        vo.setClusterInfoId(relationDO.getClusterInfoId());
+        vo.setUpdateTime(relationDO.getUpdateTime());
+        return vo;
+    }
+}

src/main/java/com/xuxd/kafka/console/beans/vo/QuerySendStatisticsVO.java

@@ -0,0 +1,36 @@
+package com.xuxd.kafka.console.beans.vo;
+
+import lombok.Data;
+
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.Map;
+
+/**
+ * @author: xuxd
+ * @since: 2023/12/1 17:49
+ **/
+@Data
+public class QuerySendStatisticsVO {
+
+    private static final String FORMAT = "yyyy-MM-dd'T'HH:mm:ss.SSSXXX";
+
+    private static final DateFormat DATE_FORMAT = new SimpleDateFormat(FORMAT);
+
+    private String topic;
+
+    private Long total;
+
+    private Map<Integer, Long> detail;
+
+    private String startTime;
+
+    private String endTime;
+
+    private String searchTime = format(new Date());
+
+    public static String format(Date date) {
+        return DATE_FORMAT.format(date);
+    }
+}

src/main/java/com/xuxd/kafka/console/config/AuthConfig.java

@@ -13,9 +13,42 @@ import org.springframework.context.annotation.Configuration;
 @ConfigurationProperties(prefix = "auth")
 public class AuthConfig {
 
+    /**
+     * 是否启用登录权限认证.
+     */
     private boolean enable;
 
+    /**
+     * 认证生成Jwt token用的,随便写.
+     */
     private String secret = "kafka-console-ui-default-secret";
 
+    /**
+     * token有效期，小时.
+     */
     private long expireHours;
+
+    /**
+     * 隐藏集群的属性信息，如果当前用户没有集群切换里的编辑权限，就不能看集群的属性信息，有开启ACL的集群需要开启这个.
+     * 不隐藏属性不行，开启ACL的时候，属性里需要配置认证信息，比如超管的用户名密码等，不等被普通角色看到.
+     */
+    private boolean hideClusterProperty;
+
+    /**
+     * 不要修改.与data-h2.sql里配置的一致即可.
+     */
+    private String hideClusterPropertyPerm = "op:cluster-switch:edit";
+
+    /**
+     * 是否启用集群的数据权限，如果启用，可以配置哪些角色看到哪些集群.
+     * 默认false是为了兼容老版本.
+     *
+     * @since 1.0.9
+     */
+    private boolean enableClusterAuthority;
+
+    /**
+     * 重新加载权限信息，版本升级替换jar包的时候，新版本里增加了新的权限菜单，这个设置为true.
+     */
+    private boolean reloadPermission;
 }

src/main/java/com/xuxd/kafka/console/config/LogConfig.java

@@ -0,0 +1,23 @@
+package com.xuxd.kafka.console.config;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @author: xuxd
+ * @since: 2023/8/20 20:00
+ **/
+@Configuration
+@ConfigurationProperties(prefix = "log")
+public class LogConfig {
+
+    private boolean printControllerLog = true;
+
+    public boolean isPrintControllerLog() {
+        return printControllerLog;
+    }
+
+    public void setPrintControllerLog(boolean printControllerLog) {
+        this.printControllerLog = printControllerLog;
+    }
+}

src/main/java/com/xuxd/kafka/console/controller/AclAuthController.java

@@ -1,5 +1,6 @@
 package com.xuxd.kafka.console.controller;
 
+import com.xuxd.kafka.console.aspect.annotation.ControllerLog;
 import com.xuxd.kafka.console.aspect.annotation.Permission;
 import com.xuxd.kafka.console.beans.AclEntry;
 import com.xuxd.kafka.console.beans.dto.AddAuthDTO;
@@ -46,6 +47,7 @@ public class AclAuthController {
         return aclService.getAclList(param.toEntry());
     }
 
+    @ControllerLog("增加Acl")
     @Permission({"acl:authority:add-principal", "acl:authority:add", "acl:sasl-scram:add-auth"})
     @PostMapping
     public Object addAcl(@RequestBody AddAuthDTO param) {
@@ -58,6 +60,7 @@ public class AclAuthController {
      * @param param entry.topic && entry.username must.
      * @return
      */
+    @ControllerLog("增加ProducerAcl")
     @Permission({"acl:authority:producer", "acl:sasl-scram:producer"})
     @PostMapping("/producer")
     public Object addProducerAcl(@RequestBody ProducerAuthDTO param) {
@@ -71,6 +74,7 @@ public class AclAuthController {
      * @param param entry.topic && entry.groupId entry.username must.
      * @return
      */
+    @ControllerLog("增加ConsumerAcl")
     @Permission({"acl:authority:consumer", "acl:sasl-scram:consumer"})
     @PostMapping("/consumer")
     public Object addConsumerAcl(@RequestBody ConsumerAuthDTO param) {
@@ -84,6 +88,7 @@ public class AclAuthController {
      * @param entry entry
      * @return
      */
+    @ControllerLog("删除Acl")
     @Permission({"acl:authority:clean", "acl:sasl-scram:pure"})
     @DeleteMapping
     public Object deleteAclByUser(@RequestBody AclEntry entry) {
@@ -96,6 +101,7 @@ public class AclAuthController {
      * @param param entry.username
      * @return
      */
+    @ControllerLog("删除Acl")
     @Permission({"acl:authority:clean", "acl:sasl-scram:pure"})
     @DeleteMapping("/user")
     public Object deleteAclByUser(@RequestBody DeleteAclDTO param) {
@@ -103,11 +109,12 @@ public class AclAuthController {
     }
 
     /**
-     * add producer acl.
+     * delete producer acl.
      *
      * @param param entry.topic && entry.username must.
      * @return
      */
+    @ControllerLog("删除ProducerAcl")
     @Permission({"acl:authority:clean", "acl:sasl-scram:pure"})
     @DeleteMapping("/producer")
     public Object deleteProducerAcl(@RequestBody ProducerAuthDTO param) {
@@ -116,11 +123,12 @@ public class AclAuthController {
     }
 
     /**
-     * add consumer acl.
+     * delete consumer acl.
      *
      * @param param entry.topic && entry.groupId entry.username must.
      * @return
      */
+    @ControllerLog("删除ConsumerAcl")
     @Permission({"acl:authority:clean", "acl:sasl-scram:pure"})
     @DeleteMapping("/consumer")
     public Object deleteConsumerAcl(@RequestBody ConsumerAuthDTO param) {
@@ -134,6 +142,7 @@ public class AclAuthController {
      * @param param acl principal.
      * @return true or false.
      */
+    @ControllerLog("清除Acl")
     @Permission({"acl:authority:clean", "acl:sasl-scram:pure"})
     @DeleteMapping("/clear")
     public Object clearAcl(@RequestBody DeleteAclDTO param) {

src/main/java/com/xuxd/kafka/console/controller/AclUserController.java

@@ -1,5 +1,6 @@
 package com.xuxd.kafka.console.controller;
 
+import com.xuxd.kafka.console.aspect.annotation.ControllerLog;
 import com.xuxd.kafka.console.aspect.annotation.Permission;
 import com.xuxd.kafka.console.beans.AclEntry;
 import com.xuxd.kafka.console.beans.AclUser;
@@ -33,12 +34,14 @@ public class AclUserController {
         return aclService.getUserList();
     }
 
+    @ControllerLog("增加SaslUser")
     @Permission({"acl:sasl-scram:add-update", "acl:sasl-scram:add-auth"})
     @PostMapping
     public Object addOrUpdateUser(@RequestBody AclUser user) {
         return aclService.addOrUpdateUser(user.getUsername(), user.getPassword());
     }
 
+    @ControllerLog("删除SaslUser")
     @Permission({"acl:sasl-scram:del", "acl:sasl-scram:pure"})
     @DeleteMapping
     public Object deleteUser(@RequestBody AclUser user) {
@@ -46,6 +49,7 @@ public class AclUserController {
     }
 
 
+    @ControllerLog("删除SaslUser和Acl")
     @Permission({"acl:sasl-scram:del", "acl:sasl-scram:pure"})
     @DeleteMapping("/auth")
     public Object deleteUserAndAuth(@RequestBody AclUser user) {
@@ -54,12 +58,12 @@ public class AclUserController {
 
     @Permission("acl:sasl-scram:detail")
     @GetMapping("/detail")
-    public Object getUserDetail(@RequestParam String username) {
+    public Object getUserDetail(@RequestParam("username") String username) {
         return aclService.getUserDetail(username);
     }
 
     @GetMapping("/scram")
-    public Object getSaslScramUserList(@RequestParam(required = false) String username) {
+    public Object getSaslScramUserList(@RequestParam(required = false, name = "username") String username) {
         AclEntry entry = new AclEntry();
         entry.setPrincipal(StringUtils.isNotBlank(username) ? username : null);
         return aclService.getSaslScramUserList(entry);

src/main/java/com/xuxd/kafka/console/controller/AuthController.java

@@ -33,4 +33,9 @@ public class AuthController {
     public ResponseData login(@RequestBody LoginUserDTO userDTO) {
         return authService.login(userDTO);
     }
+
+    @GetMapping("/own/data/auth")
+    public boolean ownDataAuthority() {
+        return authService.ownDataAuthority();
+    }
 }

src/main/java/com/xuxd/kafka/console/controller/ClientQuotaController.java

@@ -1,5 +1,6 @@
 package com.xuxd.kafka.console.controller;
 
+import com.xuxd.kafka.console.aspect.annotation.ControllerLog;
 import com.xuxd.kafka.console.aspect.annotation.Permission;
 import com.xuxd.kafka.console.beans.ResponseData;
 import com.xuxd.kafka.console.beans.dto.AlterClientQuotaDTO;
@@ -28,6 +29,7 @@ public class ClientQuotaController {
         return clientQuotaService.getClientQuotaConfigs(request.getTypes(), request.getNames());
     }
 
+    @ControllerLog("增加限流配额")
     @Permission({"quota:user:add", "quota:client:add", "quota:user-client:add", "quota:edit"})
     @PostMapping
     public Object alterClientQuotaConfigs(@RequestBody AlterClientQuotaDTO request) {
@@ -41,6 +43,7 @@ public class ClientQuotaController {
         return clientQuotaService.alterClientQuotaConfigs(request);
     }
 
+    @ControllerLog("删除限流配额")
     @Permission("quota:del")
     @DeleteMapping
     public Object deleteClientQuotaConfigs(@RequestBody AlterClientQuotaDTO request) {

src/main/java/com/xuxd/kafka/console/controller/ClusterController.java

@@ -1,16 +1,11 @@
 package com.xuxd.kafka.console.controller;
 
+import com.xuxd.kafka.console.aspect.annotation.ControllerLog;
 import com.xuxd.kafka.console.aspect.annotation.Permission;
 import com.xuxd.kafka.console.beans.dto.ClusterInfoDTO;
 import com.xuxd.kafka.console.service.ClusterService;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.web.bind.annotation.DeleteMapping;
+import org.springframework.web.bind.annotation.*;
-import org.springframework.web.bind.annotation.GetMapping;
-import org.springframework.web.bind.annotation.PostMapping;
-import org.springframework.web.bind.annotation.PutMapping;
-import org.springframework.web.bind.annotation.RequestBody;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.RestController;
 
 /**
  * kafka-console-ui.
@@ -30,23 +25,33 @@ public class ClusterController {
         return clusterService.getClusterInfo();
     }
 
+    @Permission({"op:cluster-switch", "user-manage:cluster-role:add"})
     @GetMapping("/info")
     public Object getClusterInfoList() {
         return clusterService.getClusterInfoList();
     }
 
+    @Permission({"user-manage:cluster-role:add"})
+    @GetMapping("/info/select")
+    public Object getClusterInfoListForSelect() {
+        return clusterService.getClusterInfoListForSelect();
+    }
+
+    @ControllerLog("增加集群信息")
     @Permission("op:cluster-switch:add")
     @PostMapping("/info")
     public Object addClusterInfo(@RequestBody ClusterInfoDTO dto) {
         return clusterService.addClusterInfo(dto.to());
     }
 
+    @ControllerLog("删除集群信息")
     @Permission("op:cluster-switch:del")
     @DeleteMapping("/info")
     public Object deleteClusterInfo(@RequestBody ClusterInfoDTO dto) {
         return clusterService.deleteClusterInfo(dto.getId());
     }
 
+    @ControllerLog("编辑集群信息")
     @Permission("op:cluster-switch:edit")
     @PutMapping("/info")
     public Object updateClusterInfo(@RequestBody ClusterInfoDTO dto) {

src/main/java/com/xuxd/kafka/console/controller/ClusterRoleRelationController.java

@@ -0,0 +1,42 @@
+package com.xuxd.kafka.console.controller;
+
+import com.xuxd.kafka.console.aspect.annotation.ControllerLog;
+import com.xuxd.kafka.console.aspect.annotation.Permission;
+import com.xuxd.kafka.console.beans.dto.ClusterRoleRelationDTO;
+import com.xuxd.kafka.console.service.ClusterRoleRelationService;
+import org.springframework.web.bind.annotation.*;
+
+/**
+ * @author: xuxd
+ * @since: 2023/8/23 22:01
+ **/
+@RestController
+@RequestMapping("/cluster-role/relation")
+public class ClusterRoleRelationController {
+
+    private final ClusterRoleRelationService service;
+
+    public ClusterRoleRelationController(ClusterRoleRelationService service) {
+        this.service = service;
+    }
+
+    @Permission("user-manage:cluster-role")
+    @GetMapping
+    public Object select() {
+        return service.select();
+    }
+
+    @ControllerLog("增加集群归属角色信息")
+    @Permission("user-manage:cluster-role:add")
+    @PostMapping
+    public Object add(@RequestBody ClusterRoleRelationDTO dto) {
+        return service.add(dto);
+    }
+
+    @ControllerLog("删除集群归属角色信息")
+    @Permission("user-manage:cluster-role:delete")
+    @DeleteMapping
+    public Object delete(@RequestParam("id") Long id) {
+        return service.delete(id);
+    }
+}

src/main/java/com/xuxd/kafka/console/controller/ConfigController.java

@@ -1,5 +1,6 @@
 package com.xuxd.kafka.console.controller;
 
+import com.xuxd.kafka.console.aspect.annotation.ControllerLog;
 import com.xuxd.kafka.console.aspect.annotation.Permission;
 import com.xuxd.kafka.console.beans.ResponseData;
 import com.xuxd.kafka.console.beans.dto.AlterConfigDTO;
@@ -48,12 +49,14 @@ public class ConfigController {
         return configService.getTopicConfig(topic);
     }
 
+    @ControllerLog("编辑topic配置")
     @Permission("topic:property-config:edit")
     @PostMapping("/topic")
     public Object setTopicConfig(@RequestBody AlterConfigDTO dto) {
         return configService.alterTopicConfig(dto.getEntity(), dto.to(), AlterType.SET);
     }
 
+    @ControllerLog("删除topic配置")
     @Permission("topic:property-config:del")
     @DeleteMapping("/topic")
     public Object deleteTopicConfig(@RequestBody AlterConfigDTO dto) {
@@ -66,12 +69,14 @@ public class ConfigController {
         return configService.getBrokerConfig(brokerId);
     }
 
+    @ControllerLog("设置broker配置")
     @Permission("cluster:edit")
     @PostMapping("/broker")
     public Object setBrokerConfig(@RequestBody AlterConfigDTO dto) {
         return configService.alterBrokerConfig(dto.getEntity(), dto.to(), AlterType.SET);
     }
 
+    @ControllerLog("编辑broker配置")
     @Permission("cluster:edit")
     @DeleteMapping("/broker")
     public Object deleteBrokerConfig(@RequestBody AlterConfigDTO dto) {
@@ -84,12 +89,14 @@ public class ConfigController {
         return configService.getBrokerLoggerConfig(brokerId);
     }
 
+    @ControllerLog("编辑broker日志配置")
     @Permission("cluster:edit")
     @PostMapping("/broker/logger")
     public Object setBrokerLoggerConfig(@RequestBody AlterConfigDTO dto) {
         return configService.alterBrokerLoggerConfig(dto.getEntity(), dto.to(), AlterType.SET);
     }
 
+    @ControllerLog("删除broker日志配置")
     @Permission("cluster:edit")
     @DeleteMapping("/broker/logger")
     public Object deleteBrokerLoggerConfig(@RequestBody AlterConfigDTO dto) {

src/main/java/com/xuxd/kafka/console/controller/ConsumerController.java

@@ -1,5 +1,6 @@
 package com.xuxd.kafka.console.controller;
 
+import com.xuxd.kafka.console.aspect.annotation.ControllerLog;
 import com.xuxd.kafka.console.aspect.annotation.Permission;
 import com.xuxd.kafka.console.beans.ResponseData;
 import com.xuxd.kafka.console.beans.dto.AddSubscriptionDTO;
@@ -43,30 +44,33 @@ public class ConsumerController {
         return consumerService.getConsumerGroupList(groupIdList, stateSet);
     }
 
+    @ControllerLog("删除消费组")
     @Permission("group:del")
     @DeleteMapping("/group")
-    public Object deleteConsumerGroup(@RequestParam String groupId) {
+    public Object deleteConsumerGroup(@RequestParam("groupId") String groupId) {
         return consumerService.deleteConsumerGroup(groupId);
     }
 
     @Permission("group:client")
     @GetMapping("/member")
-    public Object getConsumerMembers(@RequestParam String groupId) {
+    public Object getConsumerMembers(@RequestParam("groupId") String groupId) {
         return consumerService.getConsumerMembers(groupId);
     }
 
     @Permission("group:consumer-detail")
     @GetMapping("/detail")
-    public Object getConsumerDetail(@RequestParam String groupId) {
+    public Object getConsumerDetail(@RequestParam("groupId") String groupId) {
         return consumerService.getConsumerDetail(groupId);
     }
 
+    @ControllerLog("新增消费组")
     @Permission("group:add")
     @PostMapping("/subscription")
     public Object addSubscription(@RequestBody AddSubscriptionDTO subscriptionDTO) {
         return consumerService.addSubscription(subscriptionDTO.getGroupId(), subscriptionDTO.getTopic());
     }
 
+    @ControllerLog("重置消费位点")
     @Permission({"group:consumer-detail:min",
             "group:consumer-detail:last",
             "group:consumer-detail:timestamp",
@@ -114,19 +118,19 @@ public class ConsumerController {
     }
 
     @GetMapping("/topic/list")
-    public Object getSubscribeTopicList(@RequestParam String groupId) {
+    public Object getSubscribeTopicList(@RequestParam("groupId") String groupId) {
         return consumerService.getSubscribeTopicList(groupId);
     }
 
     @Permission({"topic:consumer-detail"})
     @GetMapping("/topic/subscribed")
-    public Object getTopicSubscribedByGroups(@RequestParam String topic) {
+    public Object getTopicSubscribedByGroups(@RequestParam("topic") String topic) {
         return consumerService.getTopicSubscribedByGroups(topic);
     }
 
     @Permission("group:offset-partition")
     @GetMapping("/offset/partition")
-    public Object getOffsetPartition(@RequestParam String groupId) {
+    public Object getOffsetPartition(@RequestParam("groupId") String groupId) {
         return consumerService.getOffsetPartition(groupId);
     }
 }

src/main/java/com/xuxd/kafka/console/controller/MessageController.java

@@ -1,12 +1,15 @@
 package com.xuxd.kafka.console.controller;
 
+import com.xuxd.kafka.console.aspect.annotation.ControllerLog;
 import com.xuxd.kafka.console.aspect.annotation.Permission;
 import com.xuxd.kafka.console.beans.QueryMessage;
 import com.xuxd.kafka.console.beans.ResponseData;
 import com.xuxd.kafka.console.beans.SendMessage;
 import com.xuxd.kafka.console.beans.dto.QueryMessageDTO;
+import com.xuxd.kafka.console.beans.dto.QuerySendStatisticsDTO;
 import com.xuxd.kafka.console.service.MessageService;
 import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.*;
 
@@ -48,18 +51,21 @@ public class MessageController {
         return messageService.deserializerList();
     }
 
-    @Permission("message:send")
     @PostMapping("/send")
+    @ControllerLog("在线发送消息")
+    @Permission("message:send")
     public Object send(@RequestBody SendMessage message) {
-        return messageService.send(message);
+        return messageService.sendWithHeader(message);
     }
 
+    @ControllerLog("重新发送消息")
     @Permission("message:resend")
     @PostMapping("/resend")
     public Object resend(@RequestBody SendMessage message) {
         return messageService.resend(message);
     }
 
+    @ControllerLog("在线删除消息")
     @Permission("message:del")
     @DeleteMapping
     public Object delete(@RequestBody List<QueryMessage> messages) {
@@ -68,4 +74,13 @@ public class MessageController {
         }
         return messageService.delete(messages);
     }
+
+    @Permission("message:send-statistics")
+    @PostMapping("/send/statistics")
+    public Object sendStatistics(@RequestBody QuerySendStatisticsDTO dto) {
+        if (StringUtils.isEmpty(dto.getTopic())) {
+            return ResponseData.create().failed("Topic is null");
+        }
+        return messageService.sendStatisticsByTime(dto);
+    }
 }

src/main/java/com/xuxd/kafka/console/controller/OperationController.java

@@ -1,5 +1,6 @@
 package com.xuxd.kafka.console.controller;
 
+import com.xuxd.kafka.console.aspect.annotation.ControllerLog;
 import com.xuxd.kafka.console.aspect.annotation.Permission;
 import com.xuxd.kafka.console.beans.TopicPartition;
 import com.xuxd.kafka.console.beans.dto.BrokerThrottleDTO;
@@ -24,12 +25,14 @@ public class OperationController {
     @Autowired
     private OperationService operationService;
 
+    @ControllerLog("同步消费位点")
     @PostMapping("/sync/consumer/offset")
     public Object syncConsumerOffset(@RequestBody SyncDataDTO dto) {
         dto.getProperties().put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, dto.getAddress());
         return operationService.syncConsumerOffset(dto.getGroupId(), dto.getTopic(), dto.getProperties());
     }
 
+    @ControllerLog("重新位点对齐")
     @PostMapping("/sync/min/offset/alignment")
     public Object minOffsetAlignment(@RequestBody SyncDataDTO dto) {
         dto.getProperties().put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, dto.getAddress());
@@ -41,23 +44,27 @@ public class OperationController {
         return operationService.getAlignmentList();
     }
 
+    @ControllerLog("deleteAlignment")
     @DeleteMapping("/sync/alignment")
-    public Object deleteAlignment(@RequestParam Long id) {
+    public Object deleteAlignment(@RequestParam("id") Long id) {
         return operationService.deleteAlignmentById(id);
     }
 
+    @ControllerLog("优先副本leader")
     @Permission({"topic:partition-detail:preferred", "op:replication-preferred"})
     @PostMapping("/replication/preferred")
     public Object electPreferredLeader(@RequestBody ReplicationDTO dto) {
         return operationService.electPreferredLeader(dto.getTopic(), dto.getPartition());
     }
 
+    @ControllerLog("配置同步限流")
     @Permission("op:config-throttle")
     @PostMapping("/broker/throttle")
     public Object configThrottle(@RequestBody BrokerThrottleDTO dto) {
         return operationService.configThrottle(dto.getBrokerList(), dto.getUnit().toKb(dto.getThrottle()));
     }
 
+    @ControllerLog("移除限流配置")
     @Permission("op:remove-throttle")
     @DeleteMapping("/broker/throttle")
     public Object removeThrottle(@RequestBody BrokerThrottleDTO dto) {
@@ -70,6 +77,7 @@ public class OperationController {
         return operationService.currentReassignments();
     }
 
+    @ControllerLog("取消副本重分配")
     @Permission("op:replication-update-detail:cancel")
     @DeleteMapping("/replication/reassignments")
     public Object cancelReassignment(@RequestBody TopicPartition partition) {

src/main/java/com/xuxd/kafka/console/controller/TopicController.java

@@ -1,5 +1,6 @@
 package com.xuxd.kafka.console.controller;
 
+import com.xuxd.kafka.console.aspect.annotation.ControllerLog;
 import com.xuxd.kafka.console.aspect.annotation.Permission;
 import com.xuxd.kafka.console.beans.ReplicaAssignment;
 import com.xuxd.kafka.console.beans.dto.AddPartitionDTO;
@@ -35,10 +36,11 @@ public class TopicController {
 
     @Permission("topic:load")
     @GetMapping("/list")
-    public Object getTopicList(@RequestParam(required = false) String topic, @RequestParam String type) {
+    public Object getTopicList(@RequestParam(required = false, name = "topic") String topic, @RequestParam("type") String type) {
         return topicService.getTopicList(topic, TopicType.valueOf(type.toUpperCase()));
     }
 
+    @ControllerLog("删除topic")
     @Permission({"topic:batch-del", "topic:del"})
     @DeleteMapping
     public Object deleteTopic(@RequestBody List<String> topics) {
@@ -47,16 +49,18 @@ public class TopicController {
 
     @Permission("topic:partition-detail")
     @GetMapping("/partition")
-    public Object getTopicPartitionInfo(@RequestParam String topic) {
+    public Object getTopicPartitionInfo(@RequestParam("topic") String topic) {
         return topicService.getTopicPartitionInfo(topic.trim());
     }
 
+    @ControllerLog("创建topic")
     @Permission("topic:add")
     @PostMapping("/new")
     public Object createNewTopic(@RequestBody NewTopicDTO topicDTO) {
         return topicService.createTopic(topicDTO.toNewTopic());
     }
 
+    @ControllerLog("增加topic分区")
     @Permission("topic:partition-add")
     @PostMapping("/partition/new")
     public Object addPartition(@RequestBody AddPartitionDTO partitionDTO) {
@@ -76,16 +80,18 @@ public class TopicController {
     }
 
     @GetMapping("/replica/assignment")
-    public Object getCurrentReplicaAssignment(@RequestParam String topic) {
+    public Object getCurrentReplicaAssignment(@RequestParam("topic") String topic) {
         return topicService.getCurrentReplicaAssignment(topic);
     }
 
+    @ControllerLog("更新副本")
     @Permission({"topic:replication-modify", "op:replication-reassign"})
     @PostMapping("/replica/assignment")
     public Object updateReplicaAssignment(@RequestBody ReplicaAssignment assignment) {
         return topicService.updateReplicaAssignment(assignment);
     }
 
+    @ControllerLog("配置限流")
     @Permission("topic:replication-sync-throttle")
     @PostMapping("/replica/throttle")
     public Object configThrottle(@RequestBody TopicThrottleDTO dto) {
@@ -94,7 +100,7 @@ public class TopicController {
 
     @Permission("topic:send-count")
     @GetMapping("/send/stats")
-    public Object sendStats(@RequestParam String topic) {
+    public Object sendStats(@RequestParam("topic") String topic) {
         return topicService.sendStats(topic);
     }
 }

src/main/java/com/xuxd/kafka/console/controller/UserManageController.java

@@ -73,14 +73,14 @@ public class UserManageController {
     @Permission("user-manage:role:del")
     @ControllerLog("删除角色")
     @DeleteMapping("/role")
-    public Object deleteRole(@RequestParam Long id) {
+    public Object deleteRole(@RequestParam("id") Long id) {
         return userManageService.deleteRole(id);
     }
 
     @Permission("user-manage:user:del")
     @ControllerLog("删除用户")
     @DeleteMapping("/user")
-    public Object deleteUser(@RequestParam Long id) {
+    public Object deleteUser(@RequestParam("id") Long id) {
         return userManageService.deleteUser(id);
     }
 

src/main/java/com/xuxd/kafka/console/dao/ClusterRoleRelationMapper.java

@@ -0,0 +1,15 @@
+package com.xuxd.kafka.console.dao;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.xuxd.kafka.console.beans.dos.ClusterRoleRelationDO;
+import org.apache.ibatis.annotations.Mapper;
+
+/**
+ * Cluster info and role relation.
+ *
+ * @author: xuxd
+ * @since: 2023/8/23 21:40
+ **/
+@Mapper
+public interface ClusterRoleRelationMapper extends BaseMapper<ClusterRoleRelationDO> {
+}

src/main/java/com/xuxd/kafka/console/dao/init/DataInit.java

@@ -71,6 +71,9 @@ public class DataInit implements SmartInitializingSingleton {
                 initData(connection, SqlParse.ROLE_TABLE);
             }
 
+            if (authConfig.isReloadPermission()) {
+                permissionMapper.delete(null);
+            }
             Integer permCount = permissionMapper.selectCount(null);
             if (permCount == null || permCount == 0) {
                 initData(connection, SqlParse.PERM_TABLE);

src/main/java/com/xuxd/kafka/console/dao/init/SqlParse.java

@@ -1,15 +1,12 @@
 package com.xuxd.kafka.console.dao.init;
 
-import com.google.common.io.Files;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
-import org.springframework.util.ResourceUtils;
 import scala.collection.mutable.StringBuilder;
 
-import java.io.File;
+import java.io.BufferedReader;
-import java.io.FileNotFoundException;
+import java.io.InputStream;
-import java.io.IOException;
+import java.io.InputStreamReader;
-import java.nio.charset.Charset;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -22,7 +19,7 @@ import java.util.Map;
 @Slf4j
 public class SqlParse {
 
-    private final String FILE = "classpath:db/data-h2.sql";
+    private final String FILE = "db/data-h2.sql";
 
     private final Map<String, List<String>> sqlMap = new HashMap<>();
 
@@ -37,8 +34,7 @@ public class SqlParse {
 
         String table = null;
         try {
-            File file = ResourceUtils.getFile(FILE);
+            List<String> lines = getSqlLines();
-            List<String> lines = Files.readLines(file, Charset.forName("UTF-8"));
             for (String str : lines) {
                 if (StringUtils.isNotEmpty(str)) {
                     if (str.indexOf("start--") > 0) {
@@ -61,9 +57,7 @@ public class SqlParse {
                     }
                 }
             }
-        } catch (FileNotFoundException e) {
+        } catch (Exception e) {
-            throw new RuntimeException(e);
-        } catch (IOException e) {
             throw new RuntimeException(e);
         }
     }
@@ -82,4 +76,21 @@ public class SqlParse {
     private boolean isSql(String str) {
         return StringUtils.isNotEmpty(str) && str.startsWith("insert");
     }
+
+    private List<String> getSqlLines() throws Exception {
+//        File file = ResourceUtils.getFile(FILE);
+//        List<String> lines = Files.readLines(file, Charset.forName("UTF-8"));
+        List<String> lines = new ArrayList<>();
+        try (InputStream inputStream = Thread.currentThread().getContextClassLoader().getResourceAsStream(FILE)) {
+            try (InputStreamReader inputStreamReader = new InputStreamReader(inputStream, "UTF-8")) {
+                try (BufferedReader reader = new BufferedReader(inputStreamReader)) {
+                    String line;
+                    while ((line = reader.readLine()) != null) {
+                        lines.add(line);
+                    }
+                }
+            }
+        }
+        return lines;
+    }
 }

src/main/java/com/xuxd/kafka/console/filter/AuthFilter.java

@@ -44,6 +44,10 @@ public class AuthFilter implements Filter {
         String accessToken = request.getHeader(TOKEN_HEADER);
 
         String requestURI = request.getRequestURI();
+        if (isResourceRequest(requestURI)) {
+            filterChain.doFilter(servletRequest, servletResponse);
+            return;
+        }
         if (requestURI.startsWith(AUTH_URI_PREFIX)) {
             filterChain.doFilter(servletRequest, servletResponse);
             return;
@@ -63,8 +67,12 @@ public class AuthFilter implements Filter {
         try {
             CredentialsContext.set(credentials);
             filterChain.doFilter(servletRequest, servletResponse);
-        }finally {
+        } finally {
             CredentialsContext.remove();
         }
     }
+
+    private boolean isResourceRequest(String requestURI) {
+        return requestURI.contains(".") || requestURI.equals("/");
+    }
 }

src/main/java/com/xuxd/kafka/console/service/AuthService.java

@@ -10,4 +10,6 @@ import com.xuxd.kafka.console.beans.dto.LoginUserDTO;
 public interface AuthService {
 
     ResponseData login(LoginUserDTO userDTO);
+
+    boolean ownDataAuthority();
 }

src/main/java/com/xuxd/kafka/console/service/ClusterRoleRelationService.java

@@ -0,0 +1,19 @@
+package com.xuxd.kafka.console.service;
+
+import com.xuxd.kafka.console.beans.ResponseData;
+import com.xuxd.kafka.console.beans.dto.ClusterRoleRelationDTO;
+
+/**
+ * Cluster info and role relation.
+ *
+ * @author: xuxd
+ * @since: 2023/8/23 21:42
+ **/
+public interface ClusterRoleRelationService {
+
+    ResponseData select();
+
+    ResponseData add(ClusterRoleRelationDTO dto);
+
+    ResponseData delete(Long id);
+}

src/main/java/com/xuxd/kafka/console/service/ClusterService.java

@@ -12,6 +12,8 @@ import com.xuxd.kafka.console.beans.dos.ClusterInfoDO;
 public interface ClusterService {
     ResponseData getClusterInfo();
 
+    ResponseData getClusterInfoListForSelect();
+
     ResponseData getClusterInfoList();
 
     ResponseData addClusterInfo(ClusterInfoDO infoDO);

src/main/java/com/xuxd/kafka/console/service/MessageService.java

@@ -1,6 +1,7 @@
 package com.xuxd.kafka.console.service;
 
 import com.xuxd.kafka.console.beans.QueryMessage;
+import com.xuxd.kafka.console.beans.dto.QuerySendStatisticsDTO;
 import com.xuxd.kafka.console.beans.ResponseData;
 import com.xuxd.kafka.console.beans.SendMessage;
 
@@ -24,7 +25,11 @@ public interface MessageService {
 
     ResponseData send(SendMessage message);
 
+    ResponseData sendWithHeader(SendMessage message);
+
     ResponseData resend(SendMessage message);
 
     ResponseData delete(List<QueryMessage> messages);
+
+    ResponseData sendStatisticsByTime(QuerySendStatisticsDTO request);
 }

src/main/java/com/xuxd/kafka/console/service/impl/AuthServiceImpl.java

@@ -93,4 +93,16 @@ public class AuthServiceImpl implements AuthService {
         return ResponseData.create().data(loginResult).success();
     }
 
+    @Override
+    public boolean ownDataAuthority() {
+        if (!authConfig.isEnable()) {
+            return true;
+        }
+        if (!authConfig.isEnableClusterAuthority()) {
+            return true;
+        }
+
+        return false;
+    }
+
 }

src/main/java/com/xuxd/kafka/console/service/impl/ClusterRoleRelationServiceImpl.java

@@ -0,0 +1,115 @@
+package com.xuxd.kafka.console.service.impl;
+
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.xuxd.kafka.console.beans.ResponseData;
+import com.xuxd.kafka.console.beans.dos.ClusterInfoDO;
+import com.xuxd.kafka.console.beans.dos.ClusterRoleRelationDO;
+import com.xuxd.kafka.console.beans.dos.SysRoleDO;
+import com.xuxd.kafka.console.beans.dto.ClusterRoleRelationDTO;
+import com.xuxd.kafka.console.beans.vo.ClusterRoleRelationVO;
+import com.xuxd.kafka.console.config.AuthConfig;
+import com.xuxd.kafka.console.dao.ClusterInfoMapper;
+import com.xuxd.kafka.console.dao.ClusterRoleRelationMapper;
+import com.xuxd.kafka.console.dao.SysRoleMapper;
+import com.xuxd.kafka.console.service.ClusterRoleRelationService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * @author: xuxd
+ * @since: 2023/8/23 21:50
+ **/
+@Slf4j
+@Service
+public class ClusterRoleRelationServiceImpl implements ClusterRoleRelationService {
+
+    private final ClusterRoleRelationMapper mapper;
+
+    private final SysRoleMapper roleMapper;
+
+    private final ClusterInfoMapper clusterInfoMapper;
+
+    private final AuthConfig authConfig;
+
+    public ClusterRoleRelationServiceImpl(final ClusterRoleRelationMapper mapper,
+                                          final SysRoleMapper roleMapper,
+                                          final ClusterInfoMapper clusterInfoMapper,
+                                          final AuthConfig authConfig) {
+        this.mapper = mapper;
+        this.roleMapper = roleMapper;
+        this.clusterInfoMapper = clusterInfoMapper;
+        this.authConfig = authConfig;
+    }
+
+    @Override
+    public ResponseData select() {
+        if (!authConfig.isEnableClusterAuthority()) {
+            return ResponseData.create().data(Collections.emptyList()).success();
+        }
+        List<ClusterRoleRelationDO> dos = mapper.selectList(null);
+
+        Map<Long, SysRoleDO> roleMap = roleMapper.selectList(null).stream().
+                collect(Collectors.toMap(SysRoleDO::getId, Function.identity(), (e1, e2) -> e2));
+        Map<Long, ClusterInfoDO> clusterMap = clusterInfoMapper.selectList(null).stream().
+                collect(Collectors.toMap(ClusterInfoDO::getId, Function.identity(), (e1, e2) -> e2));
+        List<ClusterRoleRelationVO> vos = dos.stream().
+                map(aDo -> {
+                    ClusterRoleRelationVO vo = ClusterRoleRelationVO.from(aDo);
+                    if (roleMap.containsKey(vo.getRoleId())) {
+                        vo.setRoleName(roleMap.get(vo.getRoleId()).getRoleName());
+                    }
+                    if (clusterMap.containsKey(vo.getClusterInfoId())) {
+                        vo.setClusterName(clusterMap.get(vo.getClusterInfoId()).getClusterName());
+                    }
+                    return vo;
+                }).collect(Collectors.toList());
+        return ResponseData.create().data(vos).success();
+    }
+
+    @Override
+    public ResponseData add(ClusterRoleRelationDTO dto) {
+        if (!authConfig.isEnableClusterAuthority()) {
+            return ResponseData.create().failed("未启用集群的数据权限管理");
+        }
+        ClusterRoleRelationDO relationDO = dto.toDO();
+        if (relationDO.getClusterInfoId() == -1L) {
+            // all insert
+            for (ClusterInfoDO clusterInfoDO : clusterInfoMapper.selectList(null)) {
+                ClusterRoleRelationDO aDo = new ClusterRoleRelationDO();
+                aDo.setRoleId(relationDO.getRoleId());
+                aDo.setClusterInfoId(clusterInfoDO.getId());
+                insertIfNotExist(aDo);
+            }
+        } else {
+            insertIfNotExist(relationDO);
+        }
+        return ResponseData.create().success();
+    }
+
+    @Override
+    public ResponseData delete(Long id) {
+        if (!authConfig.isEnableClusterAuthority()) {
+            return ResponseData.create().failed("未启用集群的数据权限管理");
+        }
+        mapper.deleteById(id);
+        return ResponseData.create().success();
+    }
+
+    private void insertIfNotExist(ClusterRoleRelationDO relationDO) {
+        QueryWrapper<ClusterRoleRelationDO> queryWrapper = new QueryWrapper<>();
+        queryWrapper.eq("role_id", relationDO.getRoleId()).
+                eq("cluster_info_id", relationDO.getClusterInfoId());
+        Integer count = mapper.selectCount(queryWrapper);
+        if (count > 0) {
+            log.info("已存在，不再增加：{}", relationDO);
+            return;
+        }
+        mapper.insert(relationDO);
+    }
+}

src/main/java/com/xuxd/kafka/console/service/impl/ClusterServiceImpl.java

@@ -3,15 +3,17 @@ package com.xuxd.kafka.console.service.impl;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.xuxd.kafka.console.beans.BrokerNode;
 import com.xuxd.kafka.console.beans.ClusterInfo;
+import com.xuxd.kafka.console.beans.Credentials;
 import com.xuxd.kafka.console.beans.ResponseData;
 import com.xuxd.kafka.console.beans.dos.ClusterInfoDO;
+import com.xuxd.kafka.console.beans.dos.ClusterRoleRelationDO;
 import com.xuxd.kafka.console.beans.vo.BrokerApiVersionVO;
 import com.xuxd.kafka.console.beans.vo.ClusterInfoVO;
+import com.xuxd.kafka.console.config.AuthConfig;
 import com.xuxd.kafka.console.dao.ClusterInfoMapper;
+import com.xuxd.kafka.console.dao.ClusterRoleRelationMapper;
+import com.xuxd.kafka.console.filter.CredentialsContext;
 import com.xuxd.kafka.console.service.ClusterService;
-
-import java.util.*;
-import java.util.stream.Collectors;
 import kafka.console.ClusterConsole;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections.CollectionUtils;
@@ -21,6 +23,9 @@ import org.apache.kafka.common.Node;
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.stereotype.Service;
 
+import java.util.*;
+import java.util.stream.Collectors;
+
 /**
  * kafka-console-ui.
  *
@@ -35,13 +40,22 @@ public class ClusterServiceImpl implements ClusterService {
 
     private final ClusterInfoMapper clusterInfoMapper;
 
-    public ClusterServiceImpl(ObjectProvider<ClusterConsole> clusterConsole,
+    private final AuthConfig authConfig;
-        ObjectProvider<ClusterInfoMapper> clusterInfoMapper) {
+
+    private final ClusterRoleRelationMapper clusterRoleRelationMapper;
+
+    public ClusterServiceImpl(final ObjectProvider<ClusterConsole> clusterConsole,
+                              final ObjectProvider<ClusterInfoMapper> clusterInfoMapper,
+                              final AuthConfig authConfig,
+                              final ClusterRoleRelationMapper clusterRoleRelationMapper) {
         this.clusterConsole = clusterConsole.getIfAvailable();
         this.clusterInfoMapper = clusterInfoMapper.getIfAvailable();
+        this.authConfig = authConfig;
+        this.clusterRoleRelationMapper = clusterRoleRelationMapper;
     }
 
-    @Override public ResponseData getClusterInfo() {
+    @Override
+    public ResponseData getClusterInfo() {
         ClusterInfo clusterInfo = clusterConsole.clusterInfo();
         Set<BrokerNode> nodes = clusterInfo.getNodes();
         if (nodes == null) {
@@ -52,27 +66,90 @@ public class ClusterServiceImpl implements ClusterService {
         return ResponseData.create().data(clusterInfo).success();
     }
 
-    @Override public ResponseData getClusterInfoList() {
+    @Override
-        return ResponseData.create().data(clusterInfoMapper.selectList(null)
+    public ResponseData getClusterInfoListForSelect() {
-            .stream().map(ClusterInfoVO::from).collect(Collectors.toList())).success();
+        return ResponseData.create().
+                data(clusterInfoMapper.selectList(null).stream().
+                        map(e -> {
+                            ClusterInfoVO vo = ClusterInfoVO.from(e);
+                            vo.setProperties(Collections.emptyList());
+                            vo.setAddress("");
+                            return vo;
+                        }).collect(Collectors.toList())).success();
     }
 
-    @Override public ResponseData addClusterInfo(ClusterInfoDO infoDO) {
+    @Override
+    public ResponseData getClusterInfoList() {
+        // 如果开启权限管理，当前用户没有集群切换->集群信息的编辑权限，隐藏集群的属性信息，避免ACL属性暴露出来
+        Credentials credentials = CredentialsContext.get();
+        boolean enableClusterAuthority = credentials != null && authConfig.isEnableClusterAuthority();
+        final Set<Long> clusterInfoIdSet = new HashSet<>();
+        if (enableClusterAuthority) {
+            List<Long> roleIdList = credentials.getRoleIdList();
+            QueryWrapper<ClusterRoleRelationDO> queryWrapper = new QueryWrapper<>();
+            queryWrapper.in("role_id", roleIdList);
+            clusterInfoIdSet.addAll(clusterRoleRelationMapper.selectList(queryWrapper).
+                    stream().map(ClusterRoleRelationDO::getClusterInfoId).
+                    collect(Collectors.toSet()));
+        }
+        return ResponseData.create().
+                data(clusterInfoMapper.selectList(null).stream().
+                        filter(e -> !enableClusterAuthority || clusterInfoIdSet.contains(e.getId())).
+                        map(e -> {
+                            ClusterInfoVO vo = ClusterInfoVO.from(e);
+                            if (credentials != null && credentials.isHideClusterProperty()) {
+                                vo.setProperties(Collections.emptyList());
+                            }
+                            return vo;
+                        }).collect(Collectors.toList())).success();
+    }
+
+    @Override
+    public ResponseData addClusterInfo(ClusterInfoDO infoDO) {
         QueryWrapper<ClusterInfoDO> queryWrapper = new QueryWrapper<>();
         queryWrapper.eq("cluster_name", infoDO.getClusterName());
         if (clusterInfoMapper.selectCount(queryWrapper) > 0) {
             return ResponseData.create().failed("cluster name exist.");
         }
         clusterInfoMapper.insert(infoDO);
+        Credentials credentials = CredentialsContext.get();
+        boolean enableClusterAuthority = credentials != null && authConfig.isEnableClusterAuthority();
+        if (enableClusterAuthority) {
+            for (Long roleId : credentials.getRoleIdList()) {
+                // 开启集群的数据权限控制，新增集群的时候必须要录入一条信息
+                QueryWrapper<ClusterRoleRelationDO> relationQueryWrapper = new QueryWrapper<>();
+                relationQueryWrapper.eq("role_id", roleId).
+                        eq("cluster_info_id", infoDO.getId());
+                Integer count = clusterRoleRelationMapper.selectCount(relationQueryWrapper);
+                if (count <= 0) {
+                    ClusterRoleRelationDO relationDO = new ClusterRoleRelationDO();
+                    relationDO.setRoleId(roleId);
+                    relationDO.setClusterInfoId(infoDO.getId());
+                    clusterRoleRelationMapper.insert(relationDO);
+                }
+            }
+        }
         return ResponseData.create().success();
     }
 
-    @Override public ResponseData deleteClusterInfo(Long id) {
+    @Override
+    public ResponseData deleteClusterInfo(Long id) {
         clusterInfoMapper.deleteById(id);
+        Credentials credentials = CredentialsContext.get();
+        boolean enableClusterAuthority = credentials != null && authConfig.isEnableClusterAuthority();
+        if (enableClusterAuthority) {
+            for (Long roleId : credentials.getRoleIdList()) {
+                // 开启集群的数据权限控制，删除集群的时候必须要删除对应的数据权限
+                QueryWrapper<ClusterRoleRelationDO> relationQueryWrapper = new QueryWrapper<>();
+                relationQueryWrapper.eq("role_id", roleId).eq("cluster_info_id", id);
+                clusterRoleRelationMapper.delete(relationQueryWrapper);
+            }
+        }
         return ResponseData.create().success();
     }
 
-    @Override public ResponseData updateClusterInfo(ClusterInfoDO infoDO) {
+    @Override
+    public ResponseData updateClusterInfo(ClusterInfoDO infoDO) {
         if (infoDO.getProperties() == null) {
             // null 的话不更新，这个是bug，设置为空字符串解决
             infoDO.setProperties("");
@@ -81,7 +158,8 @@ public class ClusterServiceImpl implements ClusterService {
         return ResponseData.create().success();
     }
 
-    @Override public ResponseData peekClusterInfo() {
+    @Override
+    public ResponseData peekClusterInfo() {
         List<ClusterInfoDO> dos = clusterInfoMapper.selectList(null);
         if (CollectionUtils.isEmpty(dos)) {
             return ResponseData.create().failed("No Cluster Info.");
@@ -89,7 +167,8 @@ public class ClusterServiceImpl implements ClusterService {
         return ResponseData.create().data(dos.stream().findFirst().map(ClusterInfoVO::from)).success();
     }
 
-    @Override public ResponseData getBrokerApiVersionInfo() {
+    @Override
+    public ResponseData getBrokerApiVersionInfo() {
         HashMap<Node, NodeApiVersions> map = clusterConsole.listBrokerVersionInfo();
         List<BrokerApiVersionVO> list = new ArrayList<>(map.size());
         map.forEach(((node, versions) -> {

src/main/java/com/xuxd/kafka/console/service/impl/MessageServiceImpl.java

@@ -4,20 +4,13 @@ import com.xuxd.kafka.console.beans.MessageFilter;
 import com.xuxd.kafka.console.beans.QueryMessage;
 import com.xuxd.kafka.console.beans.ResponseData;
 import com.xuxd.kafka.console.beans.SendMessage;
+import com.xuxd.kafka.console.beans.dto.QuerySendStatisticsDTO;
 import com.xuxd.kafka.console.beans.enums.FilterType;
 import com.xuxd.kafka.console.beans.vo.ConsumerRecordVO;
 import com.xuxd.kafka.console.beans.vo.MessageDetailVO;
+import com.xuxd.kafka.console.beans.vo.QuerySendStatisticsVO;
 import com.xuxd.kafka.console.service.ConsumerService;
 import com.xuxd.kafka.console.service.MessageService;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.stream.Collectors;
 import kafka.console.ConsumerConsole;
 import kafka.console.MessageConsole;
 import kafka.console.TopicConsole;
@@ -29,14 +22,7 @@ import org.apache.kafka.clients.admin.TopicDescription;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.*;
-import org.apache.kafka.common.serialization.BytesDeserializer;
-import org.apache.kafka.common.serialization.Deserializer;
-import org.apache.kafka.common.serialization.DoubleDeserializer;
-import org.apache.kafka.common.serialization.FloatDeserializer;
-import org.apache.kafka.common.serialization.IntegerDeserializer;
-import org.apache.kafka.common.serialization.LongDeserializer;
-import org.apache.kafka.common.serialization.StringDeserializer;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
@@ -44,6 +30,9 @@ import org.springframework.context.ApplicationContextAware;
 import org.springframework.stereotype.Service;
 import scala.Tuple2;
 
+import java.util.*;
+import java.util.stream.Collectors;
+
 /**
  * kafka-console-ui.
  *
@@ -79,8 +68,9 @@ public class MessageServiceImpl implements MessageService, ApplicationContextAwa
 
     public static String defaultDeserializer = "String";
 
-    @Override public ResponseData searchByTime(QueryMessage queryMessage) {
+    @Override
-        int maxNums = 5000;
+    public ResponseData searchByTime(QueryMessage queryMessage) {
+        int maxNums = queryMessage.getFilterNumber() <= 0 ? 5000 : queryMessage.getFilterNumber();
 
         Object searchContent = null;
         String headerKey = null;
@@ -144,7 +134,7 @@ public class MessageServiceImpl implements MessageService, ApplicationContextAwa
         List<ConsumerRecord<byte[], byte[]>> records = tuple2._1();
         log.info("search message by time, cost time: {}", (System.currentTimeMillis() - startTime));
         List<ConsumerRecordVO> vos = records.stream().filter(record -> record.timestamp() <= queryMessage.getEndTime())
-            .map(ConsumerRecordVO::fromConsumerRecord).collect(Collectors.toList());
+                .map(ConsumerRecordVO::fromConsumerRecord).collect(Collectors.toList());
         Map<String, Object> res = new HashMap<>();
         vos = vos.subList(0, Math.min(maxNums, vos.size()));
         res.put("maxNum", maxNums);
@@ -154,13 +144,15 @@ public class MessageServiceImpl implements MessageService, ApplicationContextAwa
         return ResponseData.create().data(res).success();
     }
 
-    @Override public ResponseData searchByOffset(QueryMessage queryMessage) {
+    @Override
+    public ResponseData searchByOffset(QueryMessage queryMessage) {
         Map<TopicPartition, ConsumerRecord<byte[], byte[]>> recordMap = searchRecordByOffset(queryMessage);
 
         return ResponseData.create().data(recordMap.values().stream().map(ConsumerRecordVO::fromConsumerRecord).collect(Collectors.toList())).success();
     }
 
-    @Override public ResponseData searchDetail(QueryMessage queryMessage) {
+    @Override
+    public ResponseData searchDetail(QueryMessage queryMessage) {
         if (queryMessage.getPartition() == -1) {
             throw new IllegalArgumentException();
         }
@@ -219,16 +211,35 @@ public class MessageServiceImpl implements MessageService, ApplicationContextAwa
         return ResponseData.create().failed("Not found message detail.");
     }
 
-    @Override public ResponseData deserializerList() {
+    @Override
+    public ResponseData deserializerList() {
         return ResponseData.create().data(deserializerDict.keySet()).success();
     }
 
-    @Override public ResponseData send(SendMessage message) {
+    @Override
+    public ResponseData send(SendMessage message) {
         messageConsole.send(message.getTopic(), message.getPartition(), message.getKey(), message.getBody(), message.getNum());
         return ResponseData.create().success();
     }
 
-    @Override public ResponseData resend(SendMessage message) {
+    @Override
+    public ResponseData sendWithHeader(SendMessage message) {
+        String[] headerKeys = message.getHeaders().stream().map(SendMessage.Header::getHeaderKey).toArray(String[]::new);
+        String[] headerValues = message.getHeaders().stream().map(SendMessage.Header::getHeaderValue).toArray(String[]::new);
+//        log.info("send with header:keys{},values{}",headerKeys, headerValues);
+        Tuple2<Object, String> tuple2 = messageConsole.send(message.getTopic(),
+                message.getPartition(),
+                message.getKey(),
+                message.getBody(),
+                message.getNum(),
+                headerKeys,
+                headerValues,
+                message.isSync());
+        return (boolean) tuple2._1 ? ResponseData.create().success() : ResponseData.create().failed(tuple2._2);
+    }
+
+    @Override
+    public ResponseData resend(SendMessage message) {
         TopicPartition partition = new TopicPartition(message.getTopic(), message.getPartition());
         Map<TopicPartition, Object> offsetTable = new HashMap<>(1, 1.0f);
         offsetTable.put(partition, message.getOffset());
@@ -255,6 +266,44 @@ public class MessageServiceImpl implements MessageService, ApplicationContextAwa
         return success ? ResponseData.create().success() : ResponseData.create().failed(tuple2._2());
     }
 
+    @Override
+    public ResponseData sendStatisticsByTime(QuerySendStatisticsDTO request) {
+
+        if (request.getPartition() != null && request.getPartition().contains(-1)) {
+            request.setPartition(Collections.emptySet());
+        }
+
+        Map<Integer, Long> startOffsetMap = topicConsole.getOffsetForTimestamp(request.getTopic(), request.getStartTime().getTime()).
+                entrySet().stream().collect(Collectors.toMap(e -> e.getKey().partition(), Map.Entry::getValue, (e1, e2) -> e2));
+        Map<Integer, Long> endOffsetMap = topicConsole.getOffsetForTimestamp(request.getTopic(), request.getEndTime().getTime()).
+                entrySet().stream().collect(Collectors.toMap(e -> e.getKey().partition(), Map.Entry::getValue, (e1, e2) -> e2));
+
+        Map<Integer, Long> diffOffsetMap = endOffsetMap.entrySet().stream().
+                collect(Collectors.toMap(e -> e.getKey(),
+                        e -> Arrays.asList(e.getValue(), startOffsetMap.getOrDefault(e.getKey(), 0L)).
+                                stream().reduce((a, b) -> a - b).get()));
+
+        if (CollectionUtils.isNotEmpty(request.getPartition())) {
+            Iterator<Map.Entry<Integer, Long>> iterator = diffOffsetMap.entrySet().iterator();
+            while (iterator.hasNext()) {
+                Integer partition = iterator.next().getKey();
+                if (!request.getPartition().contains(partition)) {
+                    iterator.remove();
+                }
+            }
+        }
+
+        Long total = diffOffsetMap.values().stream().reduce(0L, (a, b) -> a + b);
+        QuerySendStatisticsVO vo = new QuerySendStatisticsVO();
+        vo.setTopic(request.getTopic());
+        vo.setTotal(total);
+        vo.setDetail(diffOffsetMap);
+        vo.setStartTime(QuerySendStatisticsVO.format(request.getStartTime()));
+        vo.setEndTime(QuerySendStatisticsVO.format(request.getEndTime()));
+
+        return ResponseData.create().data(vo).success();
+    }
+
     private Map<TopicPartition, ConsumerRecord<byte[], byte[]>> searchRecordByOffset(QueryMessage queryMessage) {
         Set<TopicPartition> partitions = getPartitions(queryMessage);
 
@@ -276,13 +325,14 @@ public class MessageServiceImpl implements MessageService, ApplicationContextAwa
                 throw new IllegalArgumentException("Can not find topic info.");
             }
             Set<TopicPartition> set = list.get(0).partitions().stream()
-                .map(tp -> new TopicPartition(queryMessage.getTopic(), tp.partition())).collect(Collectors.toSet());
+                    .map(tp -> new TopicPartition(queryMessage.getTopic(), tp.partition())).collect(Collectors.toSet());
             partitions.addAll(set);
         }
         return partitions;
     }
 
-    @Override public void setApplicationContext(ApplicationContext context) throws BeansException {
+    @Override
+    public void setApplicationContext(ApplicationContext context) throws BeansException {
         this.applicationContext = context;
     }
 }

src/main/resources/application.yml

@@ -52,4 +52,13 @@ cron:
 auth:
   enable: false
   # 登录用户token的过期时间，单位：小时
-  expire-hours: 24
+  expire-hours: 24
+  # 隐藏集群的属性信息，如果当前用户没有集群切换里的编辑权限，就不能看集群的属性信息，有开启ACL的集群需要开启这个
+  hide-cluster-property: true
+  # 是否启用集群的数据权限，如果启用，可以配置哪些角色看到哪些集群. 不启用，即使配置了也不生效，每个角色的用户都可以看到所有集群信息.
+  enable-cluster-authority: false
+  # 重新加载权限信息，版本升级替换jar包的时候，新版本里增加了新的权限菜单，这个设置为true.然后在角色列表里分配新增加的菜单权限.
+  reload-permission: true
+log:
+  # 是否打印操作日志（增加、删除、编辑）
+  print-controller-log: true

src/main/resources/db/data-h2.sql

@@ -42,6 +42,7 @@ insert into t_sys_permission(id, name,type,parent_id,permission) values(64,'在
 insert into t_sys_permission(id, name,type,parent_id,permission) values(65,'在线删除',1,61,'message:del');
 insert into t_sys_permission(id, name,type,parent_id,permission) values(66,'消息详情',1,61,'message:detail');
 insert into t_sys_permission(id, name,type,parent_id,permission) values(67,'重新发送',1,61,'message:resend');
+insert into t_sys_permission(id, name,type,parent_id,permission) values(68,'发送统计',1,61,'message:send-statistics');
 
 insert into t_sys_permission(id, name,type,parent_id,permission) values(80,'限流',0,null,'quota');
 insert into t_sys_permission(id, name,type,parent_id,permission) values(81,'用户',1,80,'quota:user');
@@ -82,6 +83,9 @@ insert into t_sys_permission(id, name,type,parent_id,permission) values(147,'保
 insert into t_sys_permission(id, name,type,parent_id,permission) values(148,'删除',1,146,'user-manage:role:del');
 insert into t_sys_permission(id, name,type,parent_id,permission) values(149,'权限列表',1,140,'user-manage:permission');
 insert into t_sys_permission(id, name,type,parent_id,permission) values(150,'个人设置',1,140,'user-manage:setting');
+insert into t_sys_permission(id, name,type,parent_id,permission) values(151,'集群权限',1,140,'user-manage:cluster-role');
+insert into t_sys_permission(id, name,type,parent_id,permission) values(152,'新增',1,151,'user-manage:cluster-role:add');
+insert into t_sys_permission(id, name,type,parent_id,permission) values(153,'删除',1,151,'user-manage:cluster-role:delete');
 
 insert into t_sys_permission(id, name,type,parent_id,permission) values(160,'运维',0,null,'op');
 insert into t_sys_permission(id, name,type,parent_id,permission) values(161,'集群切换',1,160,'op:cluster-switch');
@@ -98,8 +102,8 @@ insert into t_sys_permission(id, name,type,parent_id,permission) values(171,'取
 -- t_sys_permission end--
 
 -- t_sys_role start--
-insert into t_sys_role(id, role_name, description, permission_ids) VALUES (1,'超级管理员','超级管理员','12,13,14,22,23,24,25,26,27,28,29,30,34,35,31,32,33,42,43,44,45,46,47,48,49,50,62,63,64,65,66,67,81,82,83,84,85,86,87,88,101,102,103,104,105,106,107,108,109,110,111,112,113,114,115,141,142,143,144,145,146,147,148,149,150,161,162,163,164,165,166,167,168,169,171,170');
+insert into t_sys_role(id, role_name, description, permission_ids) VALUES (1,'超级管理员','超级管理员','12,13,14,22,23,24,25,26,27,28,29,30,34,35,31,32,33,42,43,44,45,46,47,48,49,50,62,63,64,65,66,67,68,81,82,83,84,85,86,87,88,101,102,103,104,105,106,107,108,109,110,111,112,113,114,115,141,142,143,144,145,146,147,148,149,150,151,152,153,161,162,163,164,165,166,167,168,169,171,170');
-insert into t_sys_role(id, role_name, description, permission_ids) VALUES (2,'普通管理员','普通管理员，不能更改用户信息','12,13,14,22,23,24,25,26,27,28,29,30,34,35,31,32,33,42,43,44,45,46,47,48,49,50,62,63,64,65,66,67,81,82,83,84,85,86,87,88,101,102,103,104,105,106,107,108,109,110,111,112,113,114,115,141,146,149,150,161,162,163,164,165,166,167,168,169,171,170');
+insert into t_sys_role(id, role_name, description, permission_ids) VALUES (2,'普通管理员','普通管理员，不能更改用户信息','12,13,14,22,23,24,25,26,27,28,29,30,34,35,31,32,33,42,43,44,45,46,47,48,49,50,62,63,64,65,66,67,68,81,82,83,84,85,86,87,88,101,102,103,104,105,106,107,108,109,110,111,112,113,114,115,141,146,149,150,161,162,163,164,165,166,167,168,169,171,170');
 -- insert into t_sys_role(id, role_name, description, permission_ids) VALUES (2,'访客','访客','12,13,22,26,29,32,44,45,50,62,63,81,83,85,141,146,149,150,161,163');
 -- t_sys_role end--
 

src/main/resources/db/schema-h2.sql

@@ -4,8 +4,8 @@
 CREATE TABLE IF NOT EXISTS T_KAFKA_USER
 (
     ID              IDENTITY    NOT NULL COMMENT '主键ID',
-    USERNAME        VARCHAR(50) NOT NULL DEFAULT '' COMMENT '用户名',
+    USERNAME        VARCHAR(128) NOT NULL DEFAULT '' COMMENT '用户名',
-    PASSWORD        VARCHAR(50) NOT NULL DEFAULT '' COMMENT '密码',
+    PASSWORD        VARCHAR(128) NOT NULL DEFAULT '' COMMENT '密码',
     UPDATE_TIME     TIMESTAMP   NOT NULL DEFAULT NOW() COMMENT '更新时间',
     CLUSTER_INFO_ID BIGINT      NOT NULL COMMENT '集群信息里的集群ID',
     PRIMARY KEY (ID),
@@ -65,4 +65,15 @@ CREATE TABLE IF NOT EXISTS t_sys_user
     role_ids varchar(100) DEFAULT NULL COMMENT '分配角色的ID',
     PRIMARY KEY (id),
     UNIQUE (username)
+);
+
+-- 集群数据权限与角色绑定
+CREATE TABLE IF NOT EXISTS t_cluster_role_relation
+(
+    ID              IDENTITY   NOT NULL COMMENT '主键ID',
+    ROLE_ID         bigint(20) NOT NULL COMMENT '角色ID',
+    CLUSTER_INFO_ID bigint(20) NOT NULL COMMENT '集群信息的ID',
+    UPDATE_TIME     TIMESTAMP  NOT NULL DEFAULT NOW() COMMENT '更新时间',
+    PRIMARY KEY (id),
+    UNIQUE (ROLE_ID, CLUSTER_INFO_ID)
 );

src/main/resources/logback-spring.xml

@@ -45,7 +45,12 @@
         <appender-ref ref="AsyncFileAppender"/>
     </logger>
 
-    <logger name="org.apache.kafka.clients.consumer" level="warn" additivity="false">
+    <logger name="org.apache.kafka.common" level="warn" additivity="false">
+        <appender-ref ref="CONSOLE"/>
+        <appender-ref ref="AsyncFileAppender"/>
+    </logger>
+
+    <logger name="org.apache.kafka.clients.Metadata" level="warn" additivity="false">
         <appender-ref ref="CONSOLE"/>
         <appender-ref ref="AsyncFileAppender"/>
     </logger>
@@ -59,4 +64,6 @@
         <appender-ref ref="CONSOLE"/>
         <appender-ref ref="AsyncFileAppender"/>
     </logger>
+
+    <logger name="org.apache.kafka" level="warn"/>
 </configuration>

src/main/scala/kafka/console/BrokerApiVersion.scala

@@ -21,7 +21,7 @@ import org.apache.kafka.common.utils.{KafkaThread, LogContext, Time}
 import org.slf4j.{Logger, LoggerFactory}
 
 import java.io.IOException
-import java.util.Properties
+import java.util.{Collections, Properties}
 import java.util.concurrent.atomic.AtomicInteger
 import java.util.concurrent.{ConcurrentLinkedQueue, TimeUnit}
 import scala.jdk.CollectionConverters.{ListHasAsScala, MapHasAsJava, PropertiesHasAsScala, SetHasAsScala}
@@ -162,7 +162,7 @@ object BrokerApiVersion{
 
         def listAllBrokerVersionInfo(): Map[Node, Try[NodeApiVersions]] =
             findAllBrokers().map { broker =>
-                broker -> Try[NodeApiVersions](new NodeApiVersions(getApiVersions(broker)))
+                broker -> Try[NodeApiVersions](new NodeApiVersions(getApiVersions(broker), Collections.emptyList(), false))
             }.toMap
 
         def close(): Unit = {

src/main/scala/kafka/console/ConsumerConsole.scala

@@ -94,7 +94,7 @@ class ConsumerConsole(config: KafkaConfig) extends KafkaConsole(config: KafkaCon
                         t.lag = t.logEndOffset - t.consumerOffset
                     }
                 }
-                t.lag = t.logEndOffset - t.consumerOffset
+//                t.lag = t.logEndOffset - t.consumerOffset
                 (topicPartition, t)
             }).toMap
 

src/main/scala/kafka/console/KafkaConsole.scala

@@ -189,10 +189,10 @@ object KafkaConsole {
             case (topicPartition, listOffsetsResultInfo) => topicPartition -> new OffsetAndMetadata(listOffsetsResultInfo.offset)
         }.toMap
 
-        unsuccessfulOffsetsForTimes.foreach { entry =>
+//        unsuccessfulOffsetsForTimes.foreach { entry =>
-            log.warn(s"\nWarn: Partition " + entry._1.partition() + " from topic " + entry._1.topic() +
+//            log.warn(s"\nWarn: Partition " + entry._1.partition() + " from topic " + entry._1.topic() +
-                " is empty. Falling back to latest known offset.")
+//                " is empty. Falling back to latest known offset.")
-        }
+//        }
 
         successfulLogTimestampOffsets ++ getLogEndOffsets(admin, unsuccessfulOffsetsForTimes.keySet.toSeq, timeoutMs)
     }

src/main/scala/kafka/console/MessageConsole.scala

@@ -6,13 +6,17 @@ import com.xuxd.kafka.console.config.{ContextConfigHolder, KafkaConfig}
 import org.apache.commons.lang3.StringUtils
 import org.apache.kafka.clients.admin.{DeleteRecordsOptions, RecordsToDelete}
 import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
-import org.apache.kafka.clients.producer.ProducerRecord
+import org.apache.kafka.clients.producer.{ProducerRecord, RecordMetadata}
 import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.header.Header
+import org.apache.kafka.common.header.internals.RecordHeader
 
 import java.time.Duration
 import java.util
-import java.util.{Properties}
+import java.util.Properties
+import java.util.concurrent.Future
 import scala.collection.immutable
+import scala.collection.mutable.ArrayBuffer
 import scala.jdk.CollectionConverters.{CollectionHasAsScala, MapHasAsScala, SeqHasAsJava}
 
 /**
@@ -218,7 +222,7 @@ class MessageConsole(config: KafkaConfig) extends KafkaConsole(config: KafkaConf
 
     def send(topic: String, partition: Int, key: String, value: String, num: Int): Unit = {
         withProducerAndCatchError(producer => {
-            val nullKey = if (key != null && key.trim().length() == 0) null else key
+            val nullKey = if (key != null && key.trim().isEmpty) null else key
             for (a <- 1 to num) {
                 val record = if (partition != -1) new ProducerRecord[String, String](topic, partition, nullKey, value)
                 else new ProducerRecord[String, String](topic, nullKey, value)
@@ -228,6 +232,39 @@ class MessageConsole(config: KafkaConfig) extends KafkaConsole(config: KafkaConf
 
     }
 
+    def send(topic: String,
+             partition: Int,
+             key: String,
+             value: String,
+             num: Int,
+             headerKeys: Array[String],
+             headerValues: Array[String],
+             sync: Boolean): (Boolean, String) = {
+        withProducerAndCatchError(producer => {
+            val nullKey = if (key != null && key.trim().isEmpty) null else key
+            val results = ArrayBuffer.empty[Future[RecordMetadata]]
+            for (a <- 1 to num) {
+                val record = if (partition != -1) new ProducerRecord[String, String](topic, partition, nullKey, value)
+                else new ProducerRecord[String, String](topic, nullKey, value)
+                if (!headerKeys.isEmpty && headerKeys.length == headerValues.length) {
+                    val headers: Array[Header] = headerKeys.zip(headerValues).map { case (key, value) =>
+                        new RecordHeader(key, value.getBytes())
+                    }
+                    headers.foreach(record.headers().add)
+                }
+                results += producer.send(record)
+            }
+            if (sync) {
+                results.foreach(_.get())
+            }
+            (true, "")
+        }, e => {
+            log.error("send error.", e)
+            (false, e.getMessage)
+        })
+
+    }.asInstanceOf[(Boolean, String)]
+
     def sendSync(record: ProducerRecord[Array[Byte], Array[Byte]]): (Boolean, String) = {
         withByteProducerAndCatchError(producer => {
             val metadata = producer.send(record).get()

ui/src/store/index.js

@@ -6,6 +6,7 @@ import {
   setPermissions,
   setToken,
   setUsername,
+  deleteClusterInfo,
 } from "@/utils/local-cache";
 
 Vue.use(Vuex);
@@ -38,6 +39,9 @@ export default new Vuex.Store({
       state.clusterInfo.enableSasl = enableSasl;
       setClusterInfo(clusterInfo);
     },
+    [CLUSTER.DELETE]() {
+      deleteClusterInfo();
+    },
     [AUTH.ENABLE](state, enable) {
       state.auth.enable = enable;
     },

ui/src/store/mutation-types.js

@@ -1,5 +1,6 @@
 export const CLUSTER = {
   SWITCH: "switchCluster",
+  DELETE: "deleteClusterInfo",
 };
 
 export const AUTH = {

ui/src/utils/api.js

@@ -199,6 +199,10 @@ export const KafkaClusterApi = {
     url: "/cluster/info",
     method: "get",
   },
+  getClusterInfoListForSelect: {
+    url: "/cluster/info/select",
+    method: "get",
+  },
   addClusterInfo: {
     url: "/cluster/info",
     method: "post",
@@ -292,6 +296,10 @@ export const KafkaMessageApi = {
     url: "/message",
     method: "delete",
   },
+  sendStatistics: {
+    url: "/message/send/statistics",
+    method: "post",
+  },
 };
 
 export const KafkaClientQuotaApi = {
@@ -357,4 +365,23 @@ export const AuthApi = {
     url: "/auth/login",
     method: "post",
   },
+  ownDataAuthority: {
+    url: "/auth/own/data/auth",
+    method: "get",
+  },
+};
+
+export const ClusterRoleRelationApi = {
+  select: {
+    url: "/cluster-role/relation",
+    method: "get",
+  },
+  add: {
+    url: "/cluster-role/relation",
+    method: "post",
+  },
+  delete: {
+    url: "/cluster-role/relation",
+    method: "delete",
+  },
 };

ui/src/utils/local-cache.js

@@ -4,6 +4,10 @@ export function setClusterInfo(clusterInfo) {
   localStorage.setItem(Cache.clusterInfo, JSON.stringify(clusterInfo));
 }
 
+export function deleteClusterInfo() {
+  localStorage.removeItem(Cache.clusterInfo);
+}
+
 export function getClusterInfo() {
   const str = localStorage.getItem(Cache.clusterInfo);
   return str ? JSON.parse(str) : undefined;

ui/src/utils/request.js

@@ -22,11 +22,11 @@ const errorHandler = (error) => {
       });
       Router.push({ path: "/login-page" });
     } else if (error.response.status == 403) {
-      // const data = error.response.data;
+      const data = error.response.data;
-      // notification.error({
+      notification.error({
-      //   message: error.response.status,
+        message: error.response.status,
-      //   description: data.msg,
+        description: data.msg,
-      // });
+      });
     } else {
       const data = error.response.data;
       notification.error({

ui/src/views/acl/UpdateUser.vue

@@ -17,7 +17,7 @@
           <a-input
             placeholder="username"
             :allowClear="true"
-            :maxLength="30"
+            :maxLength="100"
             v-decorator="[
               'username',
               {
@@ -30,7 +30,7 @@
           <a-input
             placeholder="password"
             :allowClear="true"
-            :maxLength="30"
+            :maxLength="100"
             v-decorator="[
               'password',
               {

ui/src/views/group/ConsumerDetail.vue

@@ -6,7 +6,7 @@
     :mask="false"
     :destroyOnClose="true"
     :footer="null"
-    :maskClosable="false"
+    :maskClosable="true"
     @cancel="handleCancel"
   >
     <div>

ui/src/views/group/Group.vue

@@ -160,6 +160,7 @@ import Member from "@/views/group/Member";
 import ConsumerDetail from "@/views/group/ConsumerDetail";
 import AddSupscription from "@/views/group/AddSupscription";
 import OffsetTopicPartition from "@/views/group/OffsetTopicPartition";
+import { isAuthorized } from "@/utils/auth";
 
 export default {
   name: "ConsumerGroup",
@@ -235,6 +236,9 @@ export default {
       });
     },
     openConsumerMemberDialog(groupId) {
+      if (!isAuthorized("group:client")) {
+        return;
+      }
       this.showConsumerGroupDialog = true;
       this.selectDetail.resourceName = groupId;
     },

ui/src/views/group/Member.vue

@@ -6,7 +6,7 @@
     :mask="false"
     :destroyOnClose="true"
     :footer="null"
-    :maskClosable="false"
+    :maskClosable="true"
     @cancel="handleCancel"
   >
     <div>
@@ -125,4 +125,11 @@ const columns = [
 ];
 </script>
 
-<style scoped></style>
+<style scoped>
+ul ol {
+  padding-inline-start: 0px;
+}
+.ant-table-row-cell-break-word ul {
+  padding-inline-start: 0px;
+}
+</style>

ui/src/views/group/OffsetTopicPartition.vue

@@ -6,7 +6,7 @@
     :mask="false"
     :destroyOnClose="true"
     :footer="null"
-    :maskClosable="false"
+    :maskClosable="true"
     @cancel="handleCancel"
   >
     <div>

ui/src/views/login/Login.vue

@@ -9,6 +9,7 @@
     <h3 class="login-title">登录kafka-console-ui</h3>
     <a-form-item label="账号">
       <a-input
+        @keyup.enter="handleSubmit"
         style="width: 200px"
         allowClear
         v-decorator="[
@@ -20,6 +21,7 @@
     </a-form-item>
     <a-form-item label="密码">
       <a-input-password
+        @keyup.enter="handleSubmit"
         style="width: 200px"
         v-decorator="[
           'password',
@@ -28,7 +30,11 @@
       />
     </a-form-item>
     <a-form-item :wrapper-col="{ span: 16, offset: 5 }">
-      <a-button type="primary" @click="handleSubmit" :loading="loading"
+      <a-button
+        type="primary"
+        @click="handleSubmit"
+        :loading="loading"
+        @keyup.enter="handleSubmit"
         >登录</a-button
       >
     </a-form-item>
@@ -40,7 +46,7 @@ import request from "@/utils/request";
 import { AuthApi } from "@/utils/api";
 import notification from "ant-design-vue/lib/notification";
 import { mapMutations } from "vuex";
-import { AUTH } from "@/store/mutation-types";
+import { AUTH, CLUSTER } from "@/store/mutation-types";
 
 export default {
   name: "Login",
@@ -82,8 +88,19 @@ export default {
       setToken: AUTH.SET_TOKEN,
       setUsername: AUTH.SET_USERNAME,
       setPermissions: AUTH.SET_PERMISSIONS,
+      deleteClusterInfo: CLUSTER.DELETE,
     }),
   },
+  created() {
+    request({
+      url: AuthApi.ownDataAuthority.url,
+      method: AuthApi.ownDataAuthority.method,
+    }).then((res) => {
+      if (!res) {
+        this.deleteClusterInfo();
+      }
+    });
+  },
 };
 </script>
 

ui/src/views/message/Message.vue

@@ -23,6 +23,14 @@
         <a-tab-pane key="4" tab="在线删除" v-if="isAuthorized('message:del')">
           <DeleteMessage :topic-list="topicList"></DeleteMessage>
         </a-tab-pane>
+
+        <a-tab-pane
+          key="5"
+          tab="发送统计"
+          v-if="isAuthorized('message:send-statistics')"
+        >
+          <SendStatistics :topic-list="topicList"></SendStatistics>
+        </a-tab-pane>
       </a-tabs>
     </a-spin>
   </div>
@@ -31,6 +39,7 @@
 <script>
 import SearchByTime from "@/views/message/SearchByTime";
 import SearchByOffset from "@/views/message/SearchByOffset";
+import SendStatistics from "@/views/message/SendStatistics.vue";
 import request from "@/utils/request";
 import { KafkaTopicApi } from "@/utils/api";
 import notification from "ant-design-vue/lib/notification";
@@ -39,7 +48,13 @@ import DeleteMessage from "./DeleteMessage";
 import { isAuthorized, isUnauthorized } from "@/utils/auth";
 export default {
   name: "Message",
-  components: { DeleteMessage, SearchByTime, SearchByOffset, SendMessage },
+  components: {
+    DeleteMessage,
+    SearchByTime,
+    SearchByOffset,
+    SendMessage,
+    SendStatistics,
+  },
   data() {
     return {
       loading: false,

ui/src/views/message/SearchByTime.vue

@@ -60,6 +60,33 @@
             </a-col>
           </a-row>
           <hr class="hr" />
+          <a-row :gutter="24">
+            <a-col :span="24">
+              <a-form-item label="最大检索数">
+                <a-input-number
+                  v-decorator="[
+                    'filterNumber',
+                    {
+                      initialValue: 5000,
+                      rules: [
+                        {
+                          required: true,
+                          message: '输入消息数!',
+                        },
+                      ],
+                    },
+                  ]"
+                  :min="1"
+                  :max="100000"
+                />
+                <span
+                  >条
+                  注意：这里允许最多检索10万条，但是不建议将该值设置过大，这意味着一次查询要在内存里缓存这么多的数据，可能导致内存溢出；并且更大的消息量会导致更长的检索时间</span
+                >
+              </a-form-item>
+            </a-col>
+          </a-row>
+          <hr class="hr" />
           <a-row :gutter="24">
             <a-col :span="5">
               <a-form-item label="消息过滤">

ui/src/views/message/SendMessage.vue

@@ -1,7 +1,12 @@
 <template>
   <div class="content">
     <a-spin :spinning="loading">
-      <a-form :form="form" @submit="handleSubmit">
+      <a-form
+        :form="form"
+        :label-col="{ span: 5 }"
+        :wrapper-col="{ span: 12 }"
+        @submit="handleSubmit"
+      >
         <a-form-item label="Topic">
           <a-select
             class="topic-select"
@@ -34,11 +39,42 @@
             </a-select-option>
           </a-select>
         </a-form-item>
+        <a-form-item label="消息头">
+          <table>
+            <tbody>
+              <tr v-for="(row, index) in rows" :key="index">
+                <td class="w-30">
+                  <a-input v-model="row.headerKey" placeholder="key" />
+                </td>
+                <td class="w-60">
+                  <a-input v-model="row.headerValue" placeholder="value" />
+                </td>
+                <td>
+                  <a-button
+                    type="danger"
+                    @click="deleteRow(index)"
+                    v-show="rows.length > 1"
+                    >删除</a-button
+                  >
+                </td>
+                <td>
+                  <a-button
+                    type="primary"
+                    @click="addRow"
+                    v-show="index == rows.length - 1"
+                    >添加</a-button
+                  >
+                </td>
+              </tr>
+            </tbody>
+          </table>
+        </a-form-item>
         <a-form-item label="消息Key">
           <a-input v-decorator="['key', { initialValue: 'key' }]" />
         </a-form-item>
         <a-form-item label="消息体" has-feedback>
           <a-textarea
+            :autosize="{ minRows: 5 }"
             v-decorator="[
               'body',
               {
@@ -71,6 +107,22 @@
             :max="32"
           />
         </a-form-item>
+        <a-form-item label="发送类型">
+          <a-radio-group
+            v-decorator="[
+              'sync',
+              {
+                initialValue: 'false',
+                rules: [{ required: true, message: '请选择一个发送类型!' }],
+              },
+            ]"
+          >
+            <a-radio value="false"> 异步发送 </a-radio>
+            <a-radio value="true">
+              同步发送（发送失败，会返回错误信息）
+            </a-radio>
+          </a-radio-group>
+        </a-form-item>
         <a-form-item :wrapper-col="{ span: 12, offset: 5 }">
           <a-button type="primary" html-type="submit"> 提交 </a-button>
         </a-form-item>
@@ -86,17 +138,15 @@ import notification from "ant-design-vue/lib/notification";
 export default {
   name: "SendMessage",
   components: {},
-  props: {
+  props: {},
-    topicList: {
-      type: Array,
-    },
-  },
   data() {
     return {
       form: this.$form.createForm(this, { name: "message_send" }),
       loading: false,
       partitions: [],
       selectPartition: undefined,
+      rows: [{ headerKey: "", headerValue: "" }],
+      topicList: [],
     };
   },
   methods: {
@@ -137,12 +187,21 @@ export default {
       this.selectPartition = -1;
       this.getPartitionInfo(topic);
     },
+    addRow() {
+      if (this.rows.length < 32) {
+        this.rows.push({ HeaderKey: "", HeaderValue: "" });
+      }
+    },
+    deleteRow(index) {
+      this.rows.splice(index, 1);
+    },
     handleSubmit(e) {
       e.preventDefault();
       this.form.validateFields((err, values) => {
         if (!err) {
           const param = Object.assign({}, values, {
             partition: this.selectPartition,
+            headers: this.rows,
           });
           this.loading = true;
           request({
@@ -169,5 +228,11 @@ export default {
   },
 };
 </script>
-
+<style scoped>
-<style scoped></style>
+.w-30 {
+  width: 300px;
+}
+.w-60 {
+  width: 500px;
+}
+</style>

ui/src/views/message/SendStatistics.vue

@@ -0,0 +1,293 @@
+<template>
+  <div class="tab-content">
+    <a-spin :spinning="loading">
+      <div id="search-time-form-advanced-search">
+        <a-form
+          class="ant-advanced-search-form"
+          :form="form"
+          @submit="handleSearch"
+        >
+          <a-row>
+            <a-col :span="16">
+              <a-form-item label="topic">
+                <a-select
+                  class="topic-select"
+                  @change="handleTopicChange"
+                  show-search
+                  option-filter-prop="children"
+                  v-decorator="[
+                    'topic',
+                    {
+                      rules: [{ required: true, message: '请选择一个topic!' }],
+                    },
+                  ]"
+                  placeholder="请选择一个topic"
+                >
+                  <a-select-option v-for="v in topicList" :key="v" :value="v">
+                    {{ v }}
+                  </a-select-option>
+                </a-select>
+              </a-form-item>
+            </a-col>
+            <a-col :span="8">
+              <a-form-item label="分区">
+                <a-select
+                  class="type-select"
+                  show-search
+                  mode="multiple"
+                  option-filter-prop="children"
+                  v-model="selectPartition"
+                  placeholder="请选择分区"
+                >
+                  <a-select-option v-for="v in partitions" :key="v" :value="v">
+                    <span v-if="v == -1">全部</span> <span v-else>{{ v }}</span>
+                  </a-select-option>
+                </a-select>
+              </a-form-item>
+            </a-col>
+          </a-row>
+          <a-row :gutter="24">
+            <a-col :span="20">
+              <a-form-item label="时间">
+                <a-range-picker
+                  v-decorator="['time', rangeConfig]"
+                  format="YYYY-MM-DD HH:mm:ss.SSS"
+                  :show-time="{
+                    hideDisabledOptions: true,
+                    defaultValue: [
+                      moment('00:00:00.000', 'HH:mm:ss.SSS'),
+                      moment('23:59:59.999', 'HH:mm:ss.SSS'),
+                    ],
+                  }"
+                />
+              </a-form-item>
+            </a-col>
+            <a-col :span="2" :style="{ textAlign: 'right' }">
+              <a-form-item>
+                <a-button type="primary" html-type="submit"> 查询</a-button>
+              </a-form-item>
+            </a-col>
+          </a-row>
+        </a-form>
+      </div>
+      <div id="search-result-view">
+        <!--        <ul>-->
+        <!--          <li v-for="(item, index) in data" :key="index">-->
+        <!--            <fieldset>-->
+        <!--              <legend>-->
+        <!--                {{ item.topic }}, 时间: [{{ item.startTime }} ~-->
+        <!--                {{ item.endTime }}], 总数: {{ item.total }}, 查询时间:-->
+        <!--                {{ item.searchTime }}-->
+        <!--              </legend>-->
+
+        <!--            </fieldset>-->
+        <!--          </li>-->
+        <!--        </ul>-->
+        <a-collapse>
+          <a-collapse-panel
+            v-for="(item, index) in data"
+            :key="index"
+            :header="
+              item.topic +
+              ', 时间[' +
+              item.startTime +
+              ' ~ ' +
+              item.endTime +
+              '], 总数' +
+              item.total +
+              ', 查询时间' +
+              item.searchTime
+            "
+          >
+            <ul>
+              <li v-for="(value, key) in item.detail" :key="key">
+                分区:{{ key }}, 数量: {{ value }}
+              </li>
+            </ul>
+          </a-collapse-panel>
+        </a-collapse>
+      </div>
+    </a-spin>
+  </div>
+</template>
+
+<script>
+import request from "@/utils/request";
+import { KafkaMessageApi, KafkaTopicApi } from "@/utils/api";
+import notification from "ant-design-vue/lib/notification";
+import locale from "ant-design-vue/lib/date-picker/locale/zh_CN";
+import moment from "moment";
+
+export default {
+  name: "SendStatistics",
+  props: {
+    topicList: {
+      type: Array,
+    },
+  },
+  data() {
+    return {
+      moment,
+      locale,
+      loading: false,
+      form: this.$form.createForm(this, { name: "message_send_statistics" }),
+      partitions: [],
+      selectPartition: [],
+      rangeConfig: {
+        rules: [{ type: "array", required: true, message: "请选择时间!" }],
+      },
+      data: [],
+    };
+  },
+  methods: {
+    handleSearch(e) {
+      e.preventDefault();
+      this.form.validateFields((err, values) => {
+        if (!err) {
+          const data = Object.assign({}, values);
+          delete data.time;
+          data.startTime = values.time[0];
+          data.endTime = values.time[1];
+          data.partition = Array.isArray(this.selectPartition)
+            ? this.selectPartition
+            : [this.selectPartition];
+          this.loading = true;
+          request({
+            url: KafkaMessageApi.sendStatistics.url,
+            method: KafkaMessageApi.sendStatistics.method,
+            data: data,
+          }).then((res) => {
+            this.loading = false;
+            if (res.code == 0) {
+              this.data.splice(0, 0, res.data);
+            } else {
+              notification.error({
+                message: "error",
+                description: res.msg,
+              });
+            }
+          });
+        }
+      });
+    },
+    getPartitionInfo(topic) {
+      this.loading = true;
+      request({
+        url: KafkaTopicApi.getPartitionInfo.url + "?topic=" + topic,
+        method: KafkaTopicApi.getPartitionInfo.method,
+      }).then((res) => {
+        this.loading = false;
+        if (res.code != 0) {
+          notification.error({
+            message: "error",
+            description: res.msg,
+          });
+        } else {
+          this.partitions = res.data.map((v) => v.partition);
+          this.partitions.splice(0, 0, -1);
+        }
+      });
+    },
+    handleTopicChange(topic) {
+      this.selectPartition = -1;
+      this.getPartitionInfo(topic);
+    },
+    getCurrentTime() {
+      const date = new Date();
+      const yy = date.getFullYear();
+      const month = date.getMonth() + 1;
+      const mm = month < 10 ? "0" + month : month;
+      const day = date.getDate();
+      const dd = day < 10 ? "0" + day : day;
+      const hh = date.getHours();
+      const minutes = date.getMinutes();
+      const mf = minutes < 10 ? "0" + minutes : minutes;
+      const seconds = date.getSeconds();
+      const ss = seconds < 10 ? "0" + seconds : seconds;
+      return yy + "-" + mm + "-" + dd + " " + hh + ":" + mf + ":" + ss;
+    },
+  },
+};
+</script>
+
+<style scoped>
+.tab-content {
+  width: 100%;
+  height: 100%;
+}
+
+.ant-advanced-search-form {
+  padding: 24px;
+  background: #fbfbfb;
+  border: 1px solid #d9d9d9;
+  border-radius: 6px;
+}
+
+.ant-advanced-search-form .ant-form-item {
+  display: flex;
+}
+
+.ant-advanced-search-form .ant-form-item-control-wrapper {
+  flex: 1;
+}
+
+#components-form-topic-advanced-search .ant-form {
+  max-width: none;
+  margin-bottom: 1%;
+}
+
+#search-time-form-advanced-search .search-result-list {
+  margin-top: 16px;
+  border: 1px dashed #e9e9e9;
+  border-radius: 6px;
+  background-color: #fafafa;
+  min-height: 200px;
+  text-align: center;
+  padding-top: 80px;
+}
+.topic-select {
+  width: 500px !important;
+}
+
+.ant-calendar-picker {
+  width: 500px !important;
+}
+
+.type-select {
+  width: 150px !important;
+}
+.hint {
+  font-size: smaller;
+  color: green;
+}
+.ant-advanced-search-form {
+  padding-bottom: 0px;
+}
+.hr {
+  height: 1px;
+  border: none;
+  border-top: 1px dashed #0066cc;
+  padding-right: 10%;
+}
+#search-result-view ul {
+  list-style-type: none;
+  padding-left: 0px;
+  margin-top: 1%;
+}
+
+#search-result-view ul li {
+  margin-top: 1%;
+}
+
+#search-result-view fieldset {
+  border: 1px solid #333;
+  border-radius: 5px; /* 设置圆角 */
+}
+
+#search-result-view legend {
+  padding: 0.5em; /* 设置内边距 */
+}
+#search-result-view .ant-collapse {
+  margin-top: 1%;
+}
+</style>

ui/src/views/topic/ConsumedDetail.vue

@@ -6,7 +6,7 @@
     :mask="false"
     :destroyOnClose="true"
     :footer="null"
-    :maskClosable="false"
+    :maskClosable="true"
     @cancel="handleCancel"
   >
     <div>

ui/src/views/topic/PartitionInfo.vue

@@ -6,7 +6,7 @@
     :mask="false"
     :destroyOnClose="true"
     :footer="null"
-    :maskClosable="false"
+    :maskClosable="true"
     @cancel="handleCancel"
   >
     <div>

ui/src/views/topic/SendStats.vue

@@ -6,7 +6,7 @@
     :mask="false"
     :destroyOnClose="true"
     :footer="null"
-    :maskClosable="false"
+    :maskClosable="true"
     @cancel="handleCancel"
   >
     <div>

ui/src/views/topic/Topic.vue

@@ -1,5 +1,5 @@
 <template>
-  <div class="content">
+  <div class="content" v-action:topic:load>
     <a-spin :spinning="loading">
       <div class="topic">
         <div id="components-form-topic-advanced-search">

ui/src/views/topic/UpdateReplica.vue

@@ -91,6 +91,7 @@ export default {
       loading: false,
       form: this.$form.createForm(this, { name: "coordinated" }),
       brokerSize: 0,
+      brokerIdList: [],
       replicaNums: 0,
       defaultReplicaNums: 0,
     };
@@ -136,6 +137,8 @@ export default {
         method: KafkaClusterApi.getClusterInfo.method,
       }).then((res) => {
         this.brokerSize = res.data.nodes.length;
+        this.brokerIdList = res.data.nodes.map((o) => o.id);
+        this.brokerIdList.sort((a, b) => a - b);
       });
     },
     handleCancel() {
@@ -149,9 +152,16 @@ export default {
       if (this.data.partitions.length > 0) {
         this.data.partitions.forEach((p) => {
           if (value > p.replicas.length) {
+            let min = this.brokerIdList[0];
+            let max = this.brokerIdList[this.brokerSize - 1] + 1;
             let num = p.replicas[p.replicas.length - 1];
             for (let i = p.replicas.length; i < value; i++) {
-              p.replicas.push(++num % this.brokerSize);
+              ++num;
+              if (num < max) {
+                p.replicas.push(num);
+              } else {
+                p.replicas.push((num % max) + min);
+              }
             }
           }
           if (value < p.replicas.length) {

ui/src/views/user/ClusterRoleRelation.vue

@@ -0,0 +1,235 @@
+<template>
+  <div class="tab-content">
+    <a-spin :spinning="loading">
+      <div id="search-offset-form-advanced-search">
+        <a-form
+          class="ant-advanced-search-form"
+          :form="form"
+          @submit="handleSearch"
+        >
+          <a-row :gutter="24">
+            <a-col :span="16">
+              <a-form-item label="角色">
+                <a-input
+                  v-decorator="['roleName']"
+                  placeholder="请输入角色名!"
+                  @change="onRoleNameChange"
+                />
+              </a-form-item>
+            </a-col>
+            <a-col :span="2" :style="{ textAlign: 'right' }">
+              <a-form-item>
+                <a-button
+                  type="primary"
+                  html-type="submit"
+                  @click="handleSearch()"
+                >
+                  刷新
+                </a-button>
+              </a-form-item>
+            </a-col>
+          </a-row>
+        </a-form>
+      </div>
+      <div class="operation-row-button">
+        <a-button
+          type="primary"
+          @click="openCreateUserDialog()"
+          v-action:user-manage:user:add
+          >新增集群归属权限
+        </a-button>
+      </div>
+      <a-table
+        :columns="columns"
+        :data-source="filteredData"
+        bordered
+        row-key="id"
+      >
+        <div slot="operation" slot-scope="record">
+          <a-popconfirm
+            title="确认删除?"
+            ok-text="确认"
+            cancel-text="取消"
+            @confirm="deleteRelation(record)"
+          >
+            <a-button
+              size="small"
+              href="javascript:;"
+              class="operation-btn"
+              v-action:user-manage:user:del
+              >删除
+            </a-button>
+          </a-popconfirm>
+        </div>
+      </a-table>
+      <CreateClusterRoleRelation
+        @closeCreateClusterRoleRelationDialog="
+          closeCreateClusterRoleRelationDialog
+        "
+        :visible="showCreateClusterRoleRelationDialog"
+      ></CreateClusterRoleRelation>
+    </a-spin>
+  </div>
+</template>
+
+<script>
+import request from "@/utils/request";
+
+import notification from "ant-design-vue/lib/notification";
+import { ClusterRoleRelationApi } from "@/utils/api";
+import CreateClusterRoleRelation from "@/views/user/CreateClusterRoleRelation.vue";
+
+export default {
+  name: "ClusterRoleRelation",
+  components: { CreateClusterRoleRelation },
+  props: {
+    topicList: {
+      type: Array,
+    },
+  },
+  data() {
+    return {
+      loading: false,
+      form: this.$form.createForm(this, { name: "user" }),
+      data: [],
+      filteredData: [],
+      filterRoleName: "",
+      showCreateClusterRoleRelationDialog: false,
+      columns: [
+        {
+          title: "角色",
+          dataIndex: "roleName",
+          key: "roleName",
+        },
+        {
+          title: "集群",
+          dataIndex: "clusterName",
+          key: "clusterName",
+        },
+        {
+          title: "操作",
+          key: "operation",
+          scopedSlots: { customRender: "operation" },
+        },
+      ],
+    };
+  },
+  methods: {
+    handleSearch() {
+      this.form.validateFields((err) => {
+        if (!err) {
+          this.loading = true;
+          request({
+            url: ClusterRoleRelationApi.select.url,
+            method: ClusterRoleRelationApi.select.method,
+          }).then((res) => {
+            this.loading = false;
+            if (res.code == 0) {
+              this.data = res.data;
+              this.filter();
+            } else {
+              notification.error({
+                message: "error",
+                description: res.msg,
+              });
+            }
+          });
+        }
+      });
+    },
+    refresh() {
+      this.handleSearch();
+    },
+    filter() {
+      this.filteredData = this.data.filter(
+        (e) => e.roleName.indexOf(this.filterRoleName) != -1
+      );
+    },
+    onRoleNameChange(input) {
+      this.filterRoleName = input.target.value;
+      this.filter();
+    },
+    openCreateUserDialog() {
+      this.showCreateClusterRoleRelationDialog = true;
+    },
+    closeCreateClusterRoleRelationDialog(p) {
+      this.showCreateClusterRoleRelationDialog = false;
+      if (p.refresh) {
+        this.refresh();
+      }
+    },
+    deleteRelation(user) {
+      this.loading = true;
+      request({
+        url: ClusterRoleRelationApi.delete.url + "?id=" + user.id,
+        method: ClusterRoleRelationApi.delete.method,
+      }).then((res) => {
+        this.loading = false;
+        if (res.code == 0) {
+          this.refresh();
+        } else {
+          notification.error({
+            message: "error",
+            description: res.msg,
+          });
+        }
+      });
+    },
+  },
+  created() {
+    this.handleSearch();
+  },
+};
+</script>
+
+<style scoped>
+.tab-content {
+  width: 100%;
+  height: 100%;
+}
+
+.ant-advanced-search-form {
+  padding: 24px;
+  background: #fbfbfb;
+  border: 1px solid #d9d9d9;
+  border-radius: 6px;
+}
+
+.ant-advanced-search-form .ant-form-item {
+  display: flex;
+}
+
+.ant-advanced-search-form input {
+  width: 400px;
+}
+
+.ant-advanced-search-form .ant-form-item-control-wrapper {
+  flex: 1;
+}
+
+#components-form-topic-advanced-search .ant-form {
+  max-width: none;
+  margin-bottom: 1%;
+}
+
+#search-offset-form-advanced-search .search-result-list {
+  margin-top: 16px;
+  border: 1px dashed #e9e9e9;
+  border-radius: 6px;
+  background-color: #fafafa;
+  min-height: 200px;
+  text-align: center;
+  padding-top: 80px;
+}
+
+.operation-row-button {
+  height: 4%;
+  text-align: left;
+  margin-bottom: 5px;
+  margin-top: 5px;
+}
+
+.operation-btn {
+  margin-right: 3%;
+}
+</style>

ui/src/views/user/CreateClusterRoleRelation.vue

@@ -0,0 +1,175 @@
+<template>
+  <a-modal
+    title="新增集群归属权限"
+    :visible="show"
+    :width="800"
+    :mask="false"
+    :destroyOnClose="true"
+    :footer="null"
+    :maskClosable="false"
+    @cancel="handleCancel"
+  >
+    <div>
+      <a-spin :spinning="loading">
+        <a-form
+          :form="form"
+          :label-col="{ span: 5 }"
+          :wrapper-col="{ span: 12 }"
+          @submit="handleSubmit"
+        >
+          <a-form-item label="角色">
+            <a-select
+              show-search
+              option-filter-prop="children"
+              v-decorator="[
+                'roleId',
+                { rules: [{ required: true, message: '请选择一个角色!' }] },
+              ]"
+              placeholder="请选择一个角色"
+            >
+              <a-select-option
+                v-for="role in roles"
+                :key="role.id"
+                :value="role.id"
+              >
+                {{ role.roleName }}
+              </a-select-option>
+            </a-select>
+          </a-form-item>
+          <a-form-item label="集群">
+            <a-select
+              show-search
+              option-filter-prop="children"
+              v-decorator="[
+                'clusterInfoId',
+                { rules: [{ required: true, message: '请选择集群!' }] },
+              ]"
+              placeholder="请选择集群"
+            >
+              <a-select-option
+                v-for="clusterInfo in clusterInfoList"
+                :key="clusterInfo.id"
+                :value="clusterInfo.id"
+              >
+                {{ clusterInfo.clusterName }}
+              </a-select-option>
+            </a-select>
+          </a-form-item>
+          <a-form-item :wrapper-col="{ span: 12, offset: 5 }">
+            <a-button type="primary" html-type="submit"> 提交</a-button>
+          </a-form-item>
+        </a-form>
+      </a-spin>
+    </div>
+  </a-modal>
+</template>
+
+<script>
+import request from "@/utils/request";
+import notification from "ant-design-vue/es/notification";
+import {
+  UserManageApi,
+  KafkaClusterApi,
+  ClusterRoleRelationApi,
+} from "@/utils/api";
+
+export default {
+  name: "CreateClusterRoleRelation",
+  props: {
+    visible: {
+      type: Boolean,
+      default: false,
+    },
+  },
+  data() {
+    return {
+      show: this.visible,
+      data: [],
+      loading: false,
+      form: this.$form.createForm(this, { name: "coordinated" }),
+      roles: [],
+      clusterInfoList: [],
+    };
+  },
+  watch: {
+    visible(v) {
+      this.show = v;
+      if (this.show) {
+        this.getRoles();
+      }
+    },
+  },
+  methods: {
+    handleSubmit(e) {
+      e.preventDefault();
+      this.form.validateFields((err, values) => {
+        if (!err) {
+          this.loading = true;
+          request({
+            url: ClusterRoleRelationApi.add.url,
+            method: ClusterRoleRelationApi.add.method,
+            data: values,
+          }).then((res) => {
+            this.loading = false;
+            if (res.code == 0) {
+              this.$message.success(res.msg);
+              this.$emit("closeCreateClusterRoleRelationDialog", {
+                refresh: true,
+                data: res.data,
+              });
+            } else {
+              notification.error({
+                message: "error",
+                description: res.msg,
+              });
+            }
+          });
+        }
+      });
+    },
+    getRoles() {
+      this.loading = true;
+      request({
+        url: UserManageApi.getRole.url,
+        method: UserManageApi.getRole.method,
+      }).then((res) => {
+        this.loading = false;
+        if (res.code == 0) {
+          this.roles = res.data;
+        } else {
+          notification.error({
+            message: "error",
+            description: res.msg,
+          });
+        }
+      });
+    },
+    getClusterInfoList() {
+      request({
+        url: KafkaClusterApi.getClusterInfoListForSelect.url,
+        method: KafkaClusterApi.getClusterInfoListForSelect.method,
+      }).then((res) => {
+        if (res.code == 0) {
+          this.clusterInfoList = res.data;
+          this.clusterInfoList.splice(0, 0, { id: -1, clusterName: "全部" });
+        } else {
+          notification.error({
+            message: "error",
+            description: res.msg,
+          });
+        }
+      });
+    },
+    handleCancel() {
+      this.data = [];
+      this.$emit("closeCreateClusterRoleRelationDialog", { refresh: true });
+    },
+  },
+  created() {
+    this.getRoles();
+    this.getClusterInfoList();
+  },
+};
+</script>
+
+<style scoped></style>

ui/src/views/user/Role.vue

@@ -46,6 +46,7 @@
             </div>
             <div class="role-info" v-if="selectedRole.roleName">
               <a-form :form="form">
+                <h2>角色信息配置</h2>
                 <a-form-item label="角色名称">
                   <a-input
                     v-decorator="[
@@ -73,7 +74,8 @@
                   />
                 </a-form-item>
 
-                <a-form-item label="权限配置">
+                <a-form-item>
+                  <h2>功能权限配置</h2>
                   <div
                     v-for="(menuPermission, index) in selectedPermissions"
                     :key="index"

ui/src/views/user/User.vue

@@ -45,7 +45,11 @@
         bordered
         row-key="id"
       >
-        <div slot="operation" slot-scope="record" v-show="record.username != 'super-admin'">
+        <div
+          slot="operation"
+          slot-scope="record"
+          v-show="record.username != 'super-admin'"
+        >
           <a-popconfirm
             :title="'删除用户: ' + record.username + '？'"
             ok-text="确认"
@@ -178,7 +182,7 @@ export default {
     },
     filter() {
       this.filteredData = this.data.filter(
-        (e) => e.username.indexOf(this.filterUsername) != -1
+        (e) => e && e.username && e.username.indexOf(this.filterUsername) != -1
       );
     },
     onUsernameChange(input) {

ui/src/views/user/UserManage.vue

@@ -18,13 +18,20 @@
         </a-tab-pane>
         <a-tab-pane
           key="3"
+          tab="集群权限"
+          v-if="isAuthorized('user-manage:cluster-role')"
+        >
+          <ClusterRoleRelation></ClusterRoleRelation>
+        </a-tab-pane>
+        <a-tab-pane
+          key="4"
           tab="权限列表"
           v-if="isAuthorized('user-manage:permission')"
         >
           <Permission></Permission>
         </a-tab-pane>
         <a-tab-pane
-          key="4"
+          key="5"
           tab="个人设置"
           v-if="isAuthorized('user-manage:setting')"
         >
@@ -40,10 +47,11 @@ import Permission from "@/views/user/Permission.vue";
 import Role from "@/views/user/Role.vue";
 import User from "@/views/user/User.vue";
 import UserSetting from "@/views/user/UserSetting.vue";
+import ClusterRoleRelation from "@/views/user/ClusterRoleRelation.vue";
 import { isAuthorized } from "@/utils/auth";
 export default {
   name: "UserManage",
-  components: { Permission, Role, User, UserSetting },
+  components: { Permission, Role, User, UserSetting, ClusterRoleRelation },
   data() {
     return {
       loading: false,