Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ZQKC merged 1 commit into from
Sep 27, 2023
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion km-biz/src/main/java/com/xiaojukeji/know/streaming/km/biz/cluster/ClusterBrokersManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -22,5 +23,5 @@ public interface ClusterBrokersManager {
* @param clusterPhyId 物理集群 id
* @return 返回根据物理集群id获取到的集群对应broker状态信息
*/
ClusterBrokersStateVO getClusterPhyBrokersState(Long clusterPhyId);
Result<ClusterBrokersStateVO> getClusterPhyBrokersState(Long clusterPhyId);
}
26 changes: 19 additions & 7 deletions ...ain/java/com/xiaojukeji/know/streaming/km/biz/cluster/impl/ClusterBrokersManagerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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列表
Expand Down Expand Up @@ -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列表
Expand All @@ -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 ****************************************************/
Expand Down
2 changes: 1 addition & 1 deletion ...n/java/com/xiaojukeji/know/streaming/km/rest/api/v3/cluster/ClusterBrokersController.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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信息列表")
Expand Down