[Optimize]Connect 提交任务变更为只保存用户修改的配置,并修复JSON模式下配置展示不全(#1047) (#1158)

请不要在没有先创建Issue的情况下创建Pull Request。

## 变更的目的是什么

优化Content新增/编辑

## 简短的更新日志

- [Bugfix] 自定义的高级配置项，在JSON模式下未显示这些配置(#1045)
- [Optimize] 提交任务后只保存用户修改的配置，而不是将所有配置都保存起来，目前不论用户有没有修改配置都保存了所有的配置(#1047)

## 验证这一变化

XXXX

请遵循此清单，以帮助我们快速轻松地整合您的贡献：

* [ ] 一个 PR（Pull Request的简写）只解决一个问题，禁止一个 PR 解决多个问题；
* [ ] 确保 PR 有对应的 Issue（通常在您开始处理之前创建），除非是书写错误之类的琐碎更改不需要 Issue ；
* [ ] 格式化 PR 及 Commit-Log 的标题及内容，例如 #861 。PS：Commit-Log 需要在 Git Commit
代码时进行填写，在 GitHub 上修改不了；
* [ ] 编写足够详细的 PR 描述，以了解 PR 的作用、方式和原因；
* [ ] 编写必要的单元测试来验证您的逻辑更正。如果提交了新功能或重大更改，请记住在 test 模块中添加 integration-test；
* [ ] 确保编译通过，集成测试通过；

km-biz/src/main/java/com/xiaojukeji/know/streaming/km/biz/cluster/ClusterBrokersManager.java

@@ -2,7 +2,6 @@ package com.xiaojukeji.know.streaming.km.biz.cluster;
 
 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;
 
@@ -23,5 +22,5 @@ public interface ClusterBrokersManager {
      * @param clusterPhyId 物理集群 id
      * @return 返回根据物理集群id获取到的集群对应broker状态信息
      */
-    Result<ClusterBrokersStateVO> getClusterPhyBrokersState(Long clusterPhyId);
+    ClusterBrokersStateVO getClusterPhyBrokersState(Long clusterPhyId);
 }

km-biz/src/main/java/com/xiaojukeji/know/streaming/km/biz/cluster/impl/ClusterBrokersManagerImpl.java

@@ -12,13 +12,11 @@ import com.xiaojukeji.know.streaming.km.common.bean.entity.kafkacontroller.Kafka
 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;
@@ -28,7 +26,6 @@ import com.xiaojukeji.know.streaming.km.common.utils.ValidateUtils;
 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;
@@ -63,9 +60,6 @@ public class ClusterBrokersManagerImpl implements ClusterBrokersManager {
     @Autowired
     private KafkaJMXClient kafkaJMXClient;
 
-    @Autowired
-    private ClusterPhyService clusterPhyService;
-
     @Override
     public PaginationResult<ClusterBrokersOverviewVO> getClusterPhyBrokersOverview(Long clusterPhyId, ClusterBrokersOverviewDTO dto) {
         // 获取集群Broker列表
@@ -114,12 +108,7 @@ public class ClusterBrokersManagerImpl implements ClusterBrokersManager {
     }
 
     @Override
-    public Result<ClusterBrokersStateVO> getClusterPhyBrokersState(Long clusterPhyId) {
-        ClusterPhy clusterPhy = clusterPhyService.getClusterByCluster(clusterPhyId);
-        if (clusterPhy == null) {
-            return Result.buildFromRSAndMsg(ResultStatus.NOT_EXIST, MsgConstant.getClusterPhyNotExist(clusterPhyId));
-        }
-
+    public ClusterBrokersStateVO getClusterPhyBrokersState(Long clusterPhyId) {
         ClusterBrokersStateVO clusterBrokersStateVO = new ClusterBrokersStateVO();
 
         // 获取集群Broker列表
@@ -137,25 +126,24 @@ public class ClusterBrokersManagerImpl implements ClusterBrokersManager {
         );
 
         // 获取controller信息
-        Result<KafkaController> controllerResult = kafkaControllerService.getControllerFromKafka(clusterPhy);
+        KafkaController kafkaController = kafkaControllerService.getKafkaControllerFromDB(clusterPhyId);
 
         // 设置kafka-controller信息
         clusterBrokersStateVO.setKafkaControllerAlive(false);
-        if(null != controllerResult.getData()) {
+        if(null != kafkaController) {
             clusterBrokersStateVO.setKafkaController(
                     this.convert2KafkaControllerVO(
-                            controllerResult.getData(),
-                            brokerService.getBroker(clusterPhyId, controllerResult.getData().getBrokerId())
+                            kafkaController,
+                            brokerService.getBroker(clusterPhyId, kafkaController.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 Result.buildSuc(clusterBrokersStateVO);
+        return clusterBrokersStateVO;
     }
 
     /**************************************************** private method ****************************************************/