[Optimize]优化集群Brokers中, Controller显示存在延迟的问题 (#1162)

优化方式：从DB获取调整为从Kafka中实时获取。
didi · Sep 27, 2023 · 3b72f73 · 3b72f73
1 parent e2ad3af
commit 3b72f73
Show file tree

Hide file tree

Showing 3 changed files with 22 additions and 9 deletions.
diff --git a/km-biz/src/main/java/com/xiaojukeji/know/streaming/km/biz/cluster/ClusterBrokersManager.java b/km-biz/src/main/java/com/xiaojukeji/know/streaming/km/biz/cluster/ClusterBrokersManager.java
@@ -2,6 +2,7 @@
 
 import com.xiaojukeji.know.streaming.km.common.bean.dto.cluster.ClusterBrokersOverviewDTO;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.result.PaginationResult;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result;
 import com.xiaojukeji.know.streaming.km.common.bean.vo.cluster.res.ClusterBrokersOverviewVO;
 import com.xiaojukeji.know.streaming.km.common.bean.vo.cluster.res.ClusterBrokersStateVO;
 
@@ -22,5 +23,5 @@ public interface ClusterBrokersManager {
      * @param clusterPhyId 物理集群 id
      * @return 返回根据物理集群id获取到的集群对应broker状态信息
      */
-    ClusterBrokersStateVO getClusterPhyBrokersState(Long clusterPhyId);
+    Result<ClusterBrokersStateVO> getClusterPhyBrokersState(Long clusterPhyId);
 }
diff --git a/...ain/java/com/xiaojukeji/know/streaming/km/biz/cluster/impl/ClusterBrokersManagerImpl.java b/...ain/java/com/xiaojukeji/know/streaming/km/biz/cluster/impl/ClusterBrokersManagerImpl.java
@@ -12,11 +12,13 @@
 import com.xiaojukeji.know.streaming.km.common.bean.entity.metrics.BrokerMetrics;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.result.PaginationResult;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.result.ResultStatus;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.topic.Topic;
 import com.xiaojukeji.know.streaming.km.common.bean.vo.cluster.res.ClusterBrokersOverviewVO;
 import com.xiaojukeji.know.streaming.km.common.bean.vo.cluster.res.ClusterBrokersStateVO;
 import com.xiaojukeji.know.streaming.km.common.bean.vo.kafkacontroller.KafkaControllerVO;
 import com.xiaojukeji.know.streaming.km.common.constant.KafkaConstant;
+import com.xiaojukeji.know.streaming.km.common.constant.MsgConstant;
 import com.xiaojukeji.know.streaming.km.common.enums.SortTypeEnum;
 import com.xiaojukeji.know.streaming.km.common.enums.cluster.ClusterRunStateEnum;
 import com.xiaojukeji.know.streaming.km.common.utils.ConvertUtil;
@@ -26,6 +28,7 @@
 import com.xiaojukeji.know.streaming.km.core.service.broker.BrokerConfigService;
 import com.xiaojukeji.know.streaming.km.core.service.broker.BrokerMetricService;
 import com.xiaojukeji.know.streaming.km.core.service.broker.BrokerService;
+import com.xiaojukeji.know.streaming.km.core.service.cluster.ClusterPhyService;
 import com.xiaojukeji.know.streaming.km.core.service.kafkacontroller.KafkaControllerService;
 import com.xiaojukeji.know.streaming.km.core.service.topic.TopicService;
 import com.xiaojukeji.know.streaming.km.persistence.cache.LoadedClusterPhyCache;
@@ -60,6 +63,9 @@ public class ClusterBrokersManagerImpl implements ClusterBrokersManager {
     @Autowired
     private KafkaJMXClient kafkaJMXClient;
 
+    @Autowired
+    private ClusterPhyService clusterPhyService;
+
     @Override
     public PaginationResult<ClusterBrokersOverviewVO> getClusterPhyBrokersOverview(Long clusterPhyId, ClusterBrokersOverviewDTO dto) {
         // 获取集群Broker列表
@@ -108,7 +114,12 @@ public PaginationResult<ClusterBrokersOverviewVO> getClusterPhyBrokersOverview(L
     }
 
     @Override
-    public ClusterBrokersStateVO getClusterPhyBrokersState(Long clusterPhyId) {
+    public Result<ClusterBrokersStateVO> getClusterPhyBrokersState(Long clusterPhyId) {
+        ClusterPhy clusterPhy = clusterPhyService.getClusterByCluster(clusterPhyId);
+        if (clusterPhy == null) {
+            return Result.buildFromRSAndMsg(ResultStatus.NOT_EXIST, MsgConstant.getClusterPhyNotExist(clusterPhyId));
+        }
+
         ClusterBrokersStateVO clusterBrokersStateVO = new ClusterBrokersStateVO();
 
         // 获取集群Broker列表
@@ -126,24 +137,25 @@ public ClusterBrokersStateVO getClusterPhyBrokersState(Long clusterPhyId) {
         );
 
         // 获取controller信息
-        KafkaController kafkaController = kafkaControllerService.getKafkaControllerFromDB(clusterPhyId);
+        Result<KafkaController> controllerResult = kafkaControllerService.getControllerFromKafka(clusterPhy);
 
         // 设置kafka-controller信息
         clusterBrokersStateVO.setKafkaControllerAlive(false);
-        if(null != kafkaController) {
+        if(null != controllerResult.getData()) {
             clusterBrokersStateVO.setKafkaController(
                     this.convert2KafkaControllerVO(
-                            kafkaController,
-                            brokerService.getBroker(clusterPhyId, kafkaController.getBrokerId())
+                            controllerResult.getData(),
+                            brokerService.getBroker(clusterPhyId, controllerResult.getData().getBrokerId())
                     )
             );
             clusterBrokersStateVO.setKafkaControllerAlive(true);
         }
 
-        clusterBrokersStateVO.setConfigSimilar(brokerConfigService.countBrokerConfigDiffsFromDB(clusterPhyId, KafkaConstant.CONFIG_SIMILAR_IGNORED_CONFIG_KEY_LIST) <= 0
+        clusterBrokersStateVO.setConfigSimilar(
+                brokerConfigService.countBrokerConfigDiffsFromDB(clusterPhyId, KafkaConstant.CONFIG_SIMILAR_IGNORED_CONFIG_KEY_LIST) <= 0
         );
 
-        return clusterBrokersStateVO;
+        return Result.buildSuc(clusterBrokersStateVO);
     }
 
     /**************************************************** private method ****************************************************/

diff --git a/...n/java/com/xiaojukeji/know/streaming/km/rest/api/v3/cluster/ClusterBrokersController.java b/...n/java/com/xiaojukeji/know/streaming/km/rest/api/v3/cluster/ClusterBrokersController.java
@@ -52,7 +52,7 @@ public Result<List<BrokerMetadataVO>> getClusterPhyBrokersMetadata(@PathVariable
     @GetMapping(value = "clusters/{clusterPhyId}/brokers-state")
     @ResponseBody
     public Result<ClusterBrokersStateVO> getClusterPhyBrokersState(@PathVariable Long clusterPhyId) {
-        return Result.buildSuc(clusterBrokersManager.getClusterPhyBrokersState(clusterPhyId));
+        return clusterBrokersManager.getClusterPhyBrokersState(clusterPhyId);
     }
 
     @ApiOperation(value = "集群brokers信息列表")