8x48/KnowStreaming
1
0
Fork 0
mirror of https://github.com/didi/KnowStreaming.git synced 2025-12-24 11:52:08 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

增加过滤掉Broker的连接信息时，增加请求类型的判断

Browse Source
This commit is contained in:
zengqiao
2022-02-28 12:07:50 +08:00
parent 9f8e3373a8
commit b4c60eb910
1 changed files with 5 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/service/gateway/impl/TopicConnectionServiceImpl.java
View File

@@ -1,5 +1,6 @@
package com.xiaojukeji.kafka.manager.service.service.gateway.impl;
import com.xiaojukeji.kafka.manager.common.bizenum.KafkaClientEnum;
import com.xiaojukeji.kafka.manager.common.entity.pojo.gateway.TopicConnectionDO;
import com.xiaojukeji.kafka.manager.common.entity.ao.topic.TopicConnection;
import com.xiaojukeji.kafka.manager.common.constant.KafkaConstant;
@@ -167,8 +168,10 @@ public class TopicConnectionServiceImpl implements TopicConnectionService {
TopicConnection dto = convert2TopicConnectionDTO(connectionDO);
// 过滤掉broker的机器
if (brokerHostnameSet.contains(dto.getHostname()) || brokerHostnameSet.contains(dto.getIp())) {
// 发现消费的机器是broker, 则直接跳过. brokerHostnameSet有的集群存储的是IP
if (KafkaClientEnum.FETCH_CLIENT.getName().toLowerCase().equals(connectionDO.getType())
&& (brokerHostnameSet.contains(dto.getHostname()) || brokerHostnameSet.contains(dto.getIp()))) {
// 如果是fetch请求，并且是Broker的机器，则将数据进行过滤。
// bad-case：如果broker上部署了消费客户端，则这个消费客户端也会一并被过滤掉。
continue;
}