diff --git a/document/contact/weixin_contact.jpg b/document/contact/weixin_contact.jpg
index 140370e..4652109 100644
Binary files a/document/contact/weixin_contact.jpg and b/document/contact/weixin_contact.jpg differ
diff --git a/src/main/java/com/xuxd/kafka/console/aspect/PermissionAspect.java b/src/main/java/com/xuxd/kafka/console/aspect/PermissionAspect.java
index 07336b8..ff073a2 100644
--- a/src/main/java/com/xuxd/kafka/console/aspect/PermissionAspect.java
+++ b/src/main/java/com/xuxd/kafka/console/aspect/PermissionAspect.java
@@ -104,7 +104,9 @@ public class PermissionAspect {
         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) {
@@ -122,6 +124,7 @@ public class PermissionAspect {
         if (authConfig.isHideClusterProperty() && notFoundHideProperty) {
             credentials.setHideClusterProperty(true);
         }
+        credentials.setRoleIdList(roleIdList);
     }
 
     private Map<String, Set<String>> checkPermMap(String methodName, String[] value) {
diff --git a/src/main/java/com/xuxd/kafka/console/beans/Credentials.java b/src/main/java/com/xuxd/kafka/console/beans/Credentials.java
index 9db1581..87061ef 100644
--- a/src/main/java/com/xuxd/kafka/console/beans/Credentials.java
+++ b/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
@@ -20,6 +22,8 @@ public class Credentials {
      */
     private boolean hideClusterProperty;
 
+    private List<Long> roleIdList;
+
     public boolean isInvalid() {
         return this == INVALID;
     }
diff --git a/src/main/java/com/xuxd/kafka/console/config/AuthConfig.java b/src/main/java/com/xuxd/kafka/console/config/AuthConfig.java
index 695a8d9..e727cb1 100644
--- a/src/main/java/com/xuxd/kafka/console/config/AuthConfig.java
+++ b/src/main/java/com/xuxd/kafka/console/config/AuthConfig.java
@@ -13,13 +13,37 @@ 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;
 }
diff --git a/src/main/java/com/xuxd/kafka/console/controller/ClusterController.java b/src/main/java/com/xuxd/kafka/console/controller/ClusterController.java
index d7a6a78..3cfba7f 100644
--- a/src/main/java/com/xuxd/kafka/console/controller/ClusterController.java
+++ b/src/main/java/com/xuxd/kafka/console/controller/ClusterController.java
@@ -5,13 +5,7 @@ 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.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;
+import org.springframework.web.bind.annotation.*;
 
 /**
  * kafka-console-ui.
@@ -37,6 +31,12 @@ public class ClusterController {
         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")
diff --git a/src/main/java/com/xuxd/kafka/console/service/ClusterService.java b/src/main/java/com/xuxd/kafka/console/service/ClusterService.java
index b9472fb..2681373 100644
--- a/src/main/java/com/xuxd/kafka/console/service/ClusterService.java
+++ b/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);
diff --git a/src/main/java/com/xuxd/kafka/console/service/impl/ClusterRoleRelationServiceImpl.java b/src/main/java/com/xuxd/kafka/console/service/impl/ClusterRoleRelationServiceImpl.java
index b8012de..d610da3 100644
--- a/src/main/java/com/xuxd/kafka/console/service/impl/ClusterRoleRelationServiceImpl.java
+++ b/src/main/java/com/xuxd/kafka/console/service/impl/ClusterRoleRelationServiceImpl.java
@@ -7,6 +7,7 @@ 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;
@@ -14,6 +15,7 @@ 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;
@@ -33,16 +35,23 @@ public class ClusterRoleRelationServiceImpl implements ClusterRoleRelationServic
 
     private final ClusterInfoMapper clusterInfoMapper;
 
+    private final AuthConfig authConfig;
+
     public ClusterRoleRelationServiceImpl(final ClusterRoleRelationMapper mapper,
                                           final SysRoleMapper roleMapper,
-                                          final ClusterInfoMapper clusterInfoMapper) {
+                                          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().
@@ -65,6 +74,9 @@ public class ClusterRoleRelationServiceImpl implements ClusterRoleRelationServic
 
     @Override
     public ResponseData add(ClusterRoleRelationDTO dto) {
+        if (!authConfig.isEnableClusterAuthority()) {
+            return ResponseData.create().failed("未启用集群的数据权限管理");
+        }
         ClusterRoleRelationDO relationDO = dto.toDO();
         if (relationDO.getClusterInfoId() == -1L) {
             // all insert
@@ -82,6 +94,9 @@ public class ClusterRoleRelationServiceImpl implements ClusterRoleRelationServic
 
     @Override
     public ResponseData delete(Long id) {
+        if (!authConfig.isEnableClusterAuthority()) {
+            return ResponseData.create().failed("未启用集群的数据权限管理");
+        }
         mapper.deleteById(id);
         return ResponseData.create().success();
     }
diff --git a/src/main/java/com/xuxd/kafka/console/service/impl/ClusterServiceImpl.java b/src/main/java/com/xuxd/kafka/console/service/impl/ClusterServiceImpl.java
index 00aa58f..cf8be83 100644
--- a/src/main/java/com/xuxd/kafka/console/service/impl/ClusterServiceImpl.java
+++ b/src/main/java/com/xuxd/kafka/console/service/impl/ClusterServiceImpl.java
@@ -6,9 +6,12 @@ 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 kafka.console.ClusterConsole;
@@ -37,10 +40,18 @@ public class ClusterServiceImpl implements ClusterService {
 
     private final ClusterInfoMapper clusterInfoMapper;
 
-    public ClusterServiceImpl(ObjectProvider<ClusterConsole> clusterConsole,
-                              ObjectProvider<ClusterInfoMapper> clusterInfoMapper) {
+    private final AuthConfig authConfig;
+
+    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
@@ -55,18 +66,42 @@ public class ClusterServiceImpl implements ClusterService {
         return ResponseData.create().data(clusterInfo).success();
     }
 
+    @Override
+    public ResponseData getClusterInfoListForSelect() {
+        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 getClusterInfoList() {
         // 如果开启权限管理，当前用户没有集群切换->集群信息的编辑权限，隐藏集群的属性信息，避免ACL属性暴露出来
         Credentials credentials = CredentialsContext.get();
-        return ResponseData.create().data(clusterInfoMapper.selectList(null)
-                .stream().map(e -> {
-                    ClusterInfoVO vo = ClusterInfoVO.from(e);
-                    if (credentials != null && credentials.isHideClusterProperty()) {
-                        vo.setProperties(Collections.emptyList());
-                    }
-                    return vo;
-                }).collect(Collectors.toList())).success();
+        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
@@ -77,12 +112,39 @@ public class ClusterServiceImpl implements ClusterService {
             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) {
         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();
     }
 
diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml
index 2e39b41..e794502 100644
--- a/src/main/resources/application.yml
+++ b/src/main/resources/application.yml
@@ -50,12 +50,13 @@ cron:
 
 # 权限认证设置，设置为true，需要先登录才能访问
 auth:
-  enable: true
+  enable: false
   # 登录用户token的过期时间，单位：小时
   expire-hours: 24
   # 隐藏集群的属性信息，如果当前用户没有集群切换里的编辑权限，就不能看集群的属性信息，有开启ACL的集群需要开启这个
   hide-cluster-property: true
-
+  # 是否启用集群的数据权限，如果启用，可以配置哪些角色看到哪些集群. 不启用，即使配置了也不生效，每个角色的用户都可以看到所有集群信息.
+  enable-cluster-authority: false
 log:
   # 是否打印操作日志（增加、删除、编辑）
   print-controller-log: true
\ No newline at end of file
diff --git a/ui/src/utils/api.js b/ui/src/utils/api.js
index 659a41f..90649d5 100644
--- a/ui/src/utils/api.js
+++ b/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",
diff --git a/ui/src/views/user/CreateClusterRoleRelation.vue b/ui/src/views/user/CreateClusterRoleRelation.vue
index 0dd1346..f1a1bb9 100644
--- a/ui/src/views/user/CreateClusterRoleRelation.vue
+++ b/ui/src/views/user/CreateClusterRoleRelation.vue
@@ -146,8 +146,8 @@ export default {
     },
     getClusterInfoList() {
       request({
-        url: KafkaClusterApi.getClusterInfoList.url,
-        method: KafkaClusterApi.getClusterInfoList.method,
+        url: KafkaClusterApi.getClusterInfoListForSelect.url,
+        method: KafkaClusterApi.getClusterInfoListForSelect.method,
       }).then((res) => {
         if (res.code == 0) {
           this.clusterInfoList = res.data;