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

Merge branch 'dev' of https://github.com/didi/LogiKM into didi-dev

Browse Source
This commit is contained in:
Hu.Yue
2022-01-18 13:58:16 +08:00
parent fffc0c3add 71c52e6dd7
commit 16bd065098
137 changed files with 2237 additions and 831 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
kafka-manager-common/pom.xml
View File

@@ -21,15 +21,12 @@
<java_target_version>1.8</java_target_version>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<file_encoding>UTF-8</file_encoding>
<spring-version>5.1.3.RELEASE</spring-version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-web</artifactId>
<version>${spring-version}</version>
</dependency>
<!-- http -->
@@ -109,5 +106,11 @@
<groupId>junit</groupId>
<artifactId>junit</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<scope>compile</scope>
</dependency>
</dependencies>
</project>

6
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/constant/ApiPrefix.java
View File

@@ -20,12 +20,6 @@ public class ApiPrefix {
// open
public static final String API_V1_THIRD_PART_PREFIX = API_V1_PREFIX + "third-part/";
// 开放给OP的接口, 后续对 应的接口的集群都需要是物理集群
public static final String API_V1_THIRD_PART_OP_PREFIX = API_V1_THIRD_PART_PREFIX + "op/";
// 开放给Normal的接口, 后续对应的接口的集群，都需要是逻辑集群
public static final String API_V1_THIRD_PART_NORMAL_PREFIX = API_V1_THIRD_PART_PREFIX + "normal/";
// gateway
public static final String GATEWAY_API_V1_PREFIX = "/gateway" + API_V1_PREFIX;

2
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/constant/Constant.java
View File

@@ -9,7 +9,7 @@ public class Constant {
public static final Integer MAX_AVG_BYTES_DURATION = 10;
public static final Integer BATCH_INSERT_SIZE = 50;
public static final Integer BATCH_INSERT_SIZE = 30;
public static final Integer DEFAULT_SESSION_TIMEOUT_UNIT_MS = 30000;

4
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/constant/KafkaConstant.java
View File

@@ -17,6 +17,10 @@ public class KafkaConstant {
public static final String RETENTION_MS_KEY = "retention.ms";
public static final String EXTERNAL_KEY = "EXTERNAL";
public static final String INTERNAL_KEY = "INTERNAL";
private KafkaConstant() {
}
}

18
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/entity/ao/common/IpPortData.java Normal file
View File

@@ -0,0 +1,18 @@
package com.xiaojukeji.kafka.manager.common.entity.ao.common;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.io.Serializable;
@Data
@NoArgsConstructor
@AllArgsConstructor
public class IpPortData implements Serializable {
private static final long serialVersionUID = -428897032994630685L;
private String ip;
private String port;
}

11
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/entity/ao/config/expert/TopicExpiredConfig.java
View File

@@ -10,6 +10,8 @@ import java.util.List;
public class TopicExpiredConfig {
private Integer minExpiredDay = 30;
private String filterRegex = "";
private List<Long> ignoreClusterIdList = new ArrayList<>();
public Integer getMinExpiredDay() {
@@ -28,10 +30,19 @@ public class TopicExpiredConfig {
this.ignoreClusterIdList = ignoreClusterIdList;
}
public String getFilterRegex() {
return filterRegex;
}
public void setFilterRegex(String filterRegex) {
this.filterRegex = filterRegex;
}
@Override
public String toString() {
return "TopicExpiredConfig{" +
"minExpiredDay=" + minExpiredDay +
", filterRegex='" + filterRegex + '\'' +
", ignoreClusterIdList=" + ignoreClusterIdList +
'}';
}

10
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/entity/ao/topic/MineTopicSummary.java
View File

@@ -25,6 +25,8 @@ public class MineTopicSummary {
private Integer access;
private String description;
public Long getLogicalClusterId() {
return logicalClusterId;
}
@@ -105,6 +107,14 @@ public class MineTopicSummary {
this.access = access;
}
public String getDescription() {
return description;
}
public void setDescription(String description) {
this.description = description;
}
@Override
public String toString() {
return "MineTopicSummary{" +

25
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/entity/vo/normal/order/OrderVO.java
View File

@@ -27,8 +27,11 @@ public class OrderVO {
@ApiModelProperty(value = "工单状态, 0:待审批, 1:通过, 2:拒绝, 3:取消")
private Integer status;
@ApiModelProperty(value = "申请/审核时间")
private Date gmtTime;
@ApiModelProperty(value = "申请时间")
private Date gmtCreate;
@ApiModelProperty(value = "审核时间")
private Date gmtHandle;
public Long getId() {
return id;
@@ -70,12 +73,20 @@ public class OrderVO {
this.status = status;
}
public Date getGmtTime() {
return gmtTime;
public Date getGmtCreate() {
return gmtCreate;
}
public void setGmtTime(Date gmtTime) {
this.gmtTime = gmtTime;
public void setGmtCreate(Date gmtCreate) {
this.gmtCreate = gmtCreate;
}
public Date getGmtHandle() {
return gmtHandle;
}
public void setGmtHandle(Date gmtHandle) {
this.gmtHandle = gmtHandle;
}
public String getApplicant() {
@@ -95,7 +106,7 @@ public class OrderVO {
", applicant='" + applicant + '\'' +
", description='" + description + '\'' +
", status=" + status +
", gmtTime=" + gmtTime +
", gmtTime=" + gmtCreate +
'}';
}
}

11
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/entity/vo/normal/topic/TopicMineVO.java
View File

@@ -36,6 +36,9 @@ public class TopicMineVO {
@ApiModelProperty(value = "状态, 0:无权限, 1:可消费 2:可发送 3:可消费发送 4:可管理")
private Integer access;
@ApiModelProperty(value = "备注")
private String description;
public Long getClusterId() {
return clusterId;
}
@@ -108,6 +111,14 @@ public class TopicMineVO {
this.access = access;
}
public String getDescription() {
return description;
}
public void setDescription(String description) {
this.description = description;
}
@Override
public String toString() {
return "TopicMineVO{" +

20
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/events/RegionCreatedEvent.java Normal file
View File

@@ -0,0 +1,20 @@
package com.xiaojukeji.kafka.manager.common.events;
import com.xiaojukeji.kafka.manager.common.entity.pojo.RegionDO;
import lombok.Getter;
import org.springframework.context.ApplicationEvent;
/**
* Region创建事件
* @author zengqiao
* @date 22/01/1
*/
@Getter
public class RegionCreatedEvent extends ApplicationEvent {
private final RegionDO regionDO;
public RegionCreatedEvent(Object source, RegionDO regionDO) {
super(source);
this.regionDO = regionDO;
}
}

33
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/events/metrics/BaseMetricsCollectedEvent.java Normal file
View File

@@ -0,0 +1,33 @@
package com.xiaojukeji.kafka.manager.common.events.metrics;
import org.springframework.context.ApplicationEvent;
/**
* @author zengqiao
* @date 22/01/17
*/
public class BaseMetricsCollectedEvent extends ApplicationEvent {
/**
* 物理集群ID
*/
protected final Long physicalClusterId;
/**
* 收集时间，依据业务需要来设置，可以设置任务开始时间，也可以设置任务结束时间
*/
protected final Long collectTime;
public BaseMetricsCollectedEvent(Object source, Long physicalClusterId, Long collectTime) {
super(source);
this.physicalClusterId = physicalClusterId;
this.collectTime = collectTime;
}
public Long getPhysicalClusterId() {
return physicalClusterId;
}
public Long getCollectTime() {
return collectTime;
}
}

22
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/events/metrics/BatchBrokerMetricsCollectedEvent.java Normal file
View File

@@ -0,0 +1,22 @@
package com.xiaojukeji.kafka.manager.common.events.metrics;
import com.xiaojukeji.kafka.manager.common.entity.metrics.BrokerMetrics;
import java.util.List;
/**
* @author zengqiao
* @date 20/8/31
*/
public class BatchBrokerMetricsCollectedEvent extends BaseMetricsCollectedEvent {
private final List<BrokerMetrics> metricsList;
public BatchBrokerMetricsCollectedEvent(Object source, Long physicalClusterId, Long collectTime, List<BrokerMetrics> metricsList) {
super(source, physicalClusterId, collectTime);
this.metricsList = metricsList;
}
public List<BrokerMetrics> getMetricsList() {
return metricsList;
}
}

75
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/utils/BackoffUtils.java Normal file
View File

@@ -0,0 +1,75 @@
package com.xiaojukeji.kafka.manager.common.utils;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
public class BackoffUtils {
private BackoffUtils() {
}
/**
* 需要进行回退的事件信息
* <回退事件名，回退结束时间>
*/
private static final Map<String, Long> NEED_BACK_OFF_EVENT_MAP = new ConcurrentHashMap<>();
public static void backoff(long timeUnitMs) {
if (timeUnitMs <= 0) {
return;
}
try {
Thread.sleep(timeUnitMs);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} catch (Exception e) {
// ignore
}
}
/**
* 记录回退设置
* @param backoffEventKey 回退事件key
* @param backoffTimeUnitMs 回退时间(ms)
*/
public static void putNeedBackoffEvent(String backoffEventKey, Long backoffTimeUnitMs) {
if (backoffEventKey == null || backoffTimeUnitMs == null || backoffTimeUnitMs <= 0) {
return;
}
NEED_BACK_OFF_EVENT_MAP.put(backoffEventKey, backoffTimeUnitMs + System.currentTimeMillis());
}
/**
* 移除回退设置
* @param backoffEventKey 回退事件key
*/
public static void removeNeedBackoffEvent(String backoffEventKey) {
NEED_BACK_OFF_EVENT_MAP.remove(backoffEventKey);
}
/**
* 检查是否需要回退
* @param backoffEventKey 回退事件key
* @return
*/
public static boolean isNeedBackoff(String backoffEventKey) {
Long backoffEventEndTimeUnitMs = NEED_BACK_OFF_EVENT_MAP.get(backoffEventKey);
if (backoffEventEndTimeUnitMs == null) {
return false;
}
if (backoffEventEndTimeUnitMs > System.currentTimeMillis()) {
return true;
}
// 移除
try {
NEED_BACK_OFF_EVENT_MAP.remove(backoffEventKey, backoffEventEndTimeUnitMs);
} catch (Exception e) {
// 如果key不存在，这里可能出现NPE，不过不管什么异常都可以忽略
}
return false;
}
}

16
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/utils/factory/KafkaConsumerFactory.java
View File

@@ -1,7 +1,7 @@
package com.xiaojukeji.kafka.manager.common.utils.factory;
import com.alibaba.fastjson.JSONObject;
import com.xiaojukeji.kafka.manager.common.entity.pojo.ClusterDO;
import com.xiaojukeji.kafka.manager.common.utils.JsonUtils;
import com.xiaojukeji.kafka.manager.common.utils.ValidateUtils;
import org.apache.commons.pool2.BasePooledObjectFactory;
import org.apache.commons.pool2.PooledObject;
@@ -16,7 +16,7 @@ import java.util.Properties;
* @author zengqiao
* @date 20/8/24
*/
public class KafkaConsumerFactory extends BasePooledObjectFactory<KafkaConsumer> {
public class KafkaConsumerFactory extends BasePooledObjectFactory<KafkaConsumer<String, String>> {
private ClusterDO clusterDO;
public KafkaConsumerFactory(ClusterDO clusterDO) {
@@ -25,17 +25,17 @@ public class KafkaConsumerFactory extends BasePooledObjectFactory<KafkaConsumer>
@Override
public KafkaConsumer create() {
return new KafkaConsumer(createKafkaConsumerProperties(clusterDO));
return new KafkaConsumer<String, String>(createKafkaConsumerProperties(clusterDO));
}
@Override
public PooledObject<KafkaConsumer> wrap(KafkaConsumer obj) {
return new DefaultPooledObject<KafkaConsumer>(obj);
public PooledObject<KafkaConsumer<String, String>> wrap(KafkaConsumer<String, String> obj) {
return new DefaultPooledObject<>(obj);
}
@Override
public void destroyObject(final PooledObject<KafkaConsumer> p) throws Exception {
KafkaConsumer kafkaConsumer = p.getObject();
public void destroyObject(final PooledObject<KafkaConsumer<String, String>> p) throws Exception {
KafkaConsumer<String, String> kafkaConsumer = p.getObject();
if (ValidateUtils.isNull(kafkaConsumer)) {
return;
}
@@ -57,7 +57,7 @@ public class KafkaConsumerFactory extends BasePooledObjectFactory<KafkaConsumer>
if (ValidateUtils.isBlank(clusterDO.getSecurityProperties())) {
return properties;
}
properties.putAll(JSONObject.parseObject(clusterDO.getSecurityProperties(), Properties.class));
properties.putAll(JsonUtils.stringToObj(clusterDO.getSecurityProperties(), Properties.class));
return properties;
}
}

50
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/utils/jmx/JmxConfig.java
View File

@@ -1,5 +1,10 @@
package com.xiaojukeji.kafka.manager.common.utils.jmx;
import lombok.Data;
import lombok.ToString;
@Data
@ToString
public class JmxConfig {
/**
* 单台最大连接数
@@ -21,45 +26,8 @@ public class JmxConfig {
*/
private Boolean openSSL;
public Integer getMaxConn() {
return maxConn;
}
public void setMaxConn(Integer maxConn) {
this.maxConn = maxConn;
}
public String getUsername() {
return username;
}
public void setUsername(String username) {
this.username = username;
}
public String getPassword() {
return password;
}
public void setPassword(String password) {
this.password = password;
}
public Boolean isOpenSSL() {
return openSSL;
}
public void setOpenSSL(Boolean openSSL) {
this.openSSL = openSSL;
}
@Override
public String toString() {
return "JmxConfig{" +
"maxConn=" + maxConn +
", username='" + username + '\'' +
", password='" + password + '\'' +
", openSSL=" + openSSL +
'}';
}
/**
* 连接重试回退事件
*/
private Long retryConnectBackoffTimeUnitMs;
}

108
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/utils/jmx/JmxConnectorWrap.java
View File

@@ -1,5 +1,6 @@
package com.xiaojukeji.kafka.manager.common.utils.jmx;
import com.xiaojukeji.kafka.manager.common.utils.BackoffUtils;
import com.xiaojukeji.kafka.manager.common.utils.ValidateUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -13,11 +14,11 @@ import javax.naming.Context;
import javax.rmi.ssl.SslRMIClientSocketFactory;
import java.io.IOException;
import java.net.MalformedURLException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;
/**
* JMXConnector包装类
@@ -25,19 +26,27 @@ import java.util.concurrent.atomic.AtomicInteger;
* @date 2015/11/9.
*/
public class JmxConnectorWrap {
private final static Logger LOGGER = LoggerFactory.getLogger(JmxConnectorWrap.class);
private static final Logger LOGGER = LoggerFactory.getLogger(JmxConnectorWrap.class);
private String host;
private final Long physicalClusterId;
private int port;
private final Integer brokerId;
private final String host;
private final int port;
private JMXConnector jmxConnector;
private AtomicInteger atomicInteger;
private final AtomicInteger atomicInteger;
private JmxConfig jmxConfig;
public JmxConnectorWrap(String host, int port, JmxConfig jmxConfig) {
private final ReentrantLock modifyJMXConnectorLock = new ReentrantLock();
public JmxConnectorWrap(Long physicalClusterId, Integer brokerId, String host, int port, JmxConfig jmxConfig) {
this.physicalClusterId = physicalClusterId;
this.brokerId = brokerId;
this.host = host;
this.port = port;
this.jmxConfig = jmxConfig;
@@ -45,7 +54,12 @@ public class JmxConnectorWrap {
this.jmxConfig = new JmxConfig();
}
if (ValidateUtils.isNullOrLessThanZero(this.jmxConfig.getMaxConn())) {
this.jmxConfig.setMaxConn(1);
// 默认设置20
this.jmxConfig.setMaxConn(20);
}
if (ValidateUtils.isNullOrLessThanZero(this.jmxConfig.getRetryConnectBackoffTimeUnitMs())) {
// 默认回退10分钟
this.jmxConfig.setRetryConnectBackoffTimeUnitMs(10 * 60 * 1000L);
}
this.atomicInteger = new AtomicInteger(this.jmxConfig.getMaxConn());
}
@@ -57,17 +71,40 @@ public class JmxConnectorWrap {
if (port == -1) {
return false;
}
return createJmxConnector();
return safeCreateJmxConnector();
}
public synchronized void close() {
public void close() {
this.closeJmxConnect();
}
public void closeJmxConnect() {
if (jmxConnector == null) {
return;
}
try {
modifyJMXConnectorLock.lock();
// 移除设置的backoff事件
BackoffUtils.removeNeedBackoffEvent(buildConnectJmxFailedBackoffEventKey(physicalClusterId, brokerId));
jmxConnector.close();
} catch (IOException e) {
LOGGER.warn("close JmxConnector exception, host:{} port:{}.", host, port, e);
} catch (Exception e) {
LOGGER.error("close JmxConnector exception, physicalClusterId:{} brokerId:{} host:{} port:{}.", physicalClusterId, brokerId, host, port, e);
} finally {
jmxConnector = null;
modifyJMXConnectorLock.unlock();
}
}
private boolean safeCreateJmxConnector() {
try {
modifyJMXConnectorLock.lock();
return createJmxConnector();
} finally {
modifyJMXConnectorLock.unlock();
}
}
@@ -75,6 +112,12 @@ public class JmxConnectorWrap {
if (jmxConnector != null) {
return true;
}
if (BackoffUtils.isNeedBackoff(buildConnectJmxFailedBackoffEventKey(physicalClusterId, brokerId))) {
// 被设置了需要进行回退，则本次不进行创建
return false;
}
String jmxUrl = String.format("service:jmx:rmi:///jndi/rmi://%s:%d/jmxrmi", host, port);
try {
Map<String, Object> environment = new HashMap<String, Object>();
@@ -82,7 +125,9 @@ public class JmxConnectorWrap {
// fixed by riyuetianmu
environment.put(JMXConnector.CREDENTIALS, new String[]{this.jmxConfig.getUsername(), this.jmxConfig.getPassword()});
}
if (jmxConfig.isOpenSSL() != null && this.jmxConfig.isOpenSSL()) {
if (jmxConfig.getOpenSSL() != null && this.jmxConfig.getOpenSSL()) {
// 开启ssl
environment.put(Context.SECURITY_PROTOCOL, "ssl");
SslRMIClientSocketFactory clientSocketFactory = new SslRMIClientSocketFactory();
environment.put(RMIConnectorServer.RMI_CLIENT_SOCKET_FACTORY_ATTRIBUTE, clientSocketFactory);
@@ -90,13 +135,17 @@ public class JmxConnectorWrap {
}
jmxConnector = JMXConnectorFactory.connect(new JMXServiceURL(jmxUrl), environment);
LOGGER.info("JMX connect success, host:{} port:{}.", host, port);
LOGGER.info("connect JMX success, physicalClusterId:{} brokerId:{} host:{} port:{}.", physicalClusterId, brokerId, host, port);
return true;
} catch (MalformedURLException e) {
LOGGER.error("JMX url exception, host:{} port:{} jmxUrl:{}", host, port, jmxUrl, e);
LOGGER.error("connect JMX failed, JMX url exception, physicalClusterId:{} brokerId:{} host:{} port:{} jmxUrl:{}.", physicalClusterId, brokerId, host, port, jmxUrl, e);
} catch (Exception e) {
LOGGER.error("JMX connect exception, host:{} port:{}.", host, port, e);
LOGGER.error("connect JMX failed, physicalClusterId:{} brokerId:{} host:{} port:{}.", physicalClusterId, brokerId, host, port, e);
}
// 设置连接backoff
BackoffUtils.putNeedBackoffEvent(buildConnectJmxFailedBackoffEventKey(physicalClusterId, brokerId), this.jmxConfig.getRetryConnectBackoffTimeUnitMs());
return false;
}
@@ -110,6 +159,11 @@ public class JmxConnectorWrap {
acquire();
MBeanServerConnection mBeanServerConnection = jmxConnector.getMBeanServerConnection();
return mBeanServerConnection.getAttribute(name, attribute);
} catch (IOException ioe) {
// io错误，则重置连接
this.closeJmxConnect();
throw ioe;
} finally {
atomicInteger.incrementAndGet();
}
@@ -125,6 +179,11 @@ public class JmxConnectorWrap {
acquire();
MBeanServerConnection mBeanServerConnection = jmxConnector.getMBeanServerConnection();
return mBeanServerConnection.getAttributes(name, attributes);
} catch (IOException ioe) {
// io错误，则重置连接
this.closeJmxConnect();
throw ioe;
} finally {
atomicInteger.incrementAndGet();
}
@@ -137,6 +196,11 @@ public class JmxConnectorWrap {
acquire();
MBeanServerConnection mBeanServerConnection = jmxConnector.getMBeanServerConnection();
return mBeanServerConnection.queryNames(name, query);
} catch (IOException ioe) {
// io错误，则重置连接
this.closeJmxConnect();
throw ioe;
} finally {
atomicInteger.incrementAndGet();
}
@@ -146,19 +210,21 @@ public class JmxConnectorWrap {
long now = System.currentTimeMillis();
while (true) {
try {
if (System.currentTimeMillis() - now > 60000) {
break;
}
int num = atomicInteger.get();
if (num <= 0) {
Thread.sleep(2);
continue;
BackoffUtils.backoff(2);
}
if (atomicInteger.compareAndSet(num, num - 1)) {
if (atomicInteger.compareAndSet(num, num - 1) || System.currentTimeMillis() - now > 6000) {
break;
}
} catch (Exception e) {
// ignore
}
}
}
private static String buildConnectJmxFailedBackoffEventKey(Long physicalClusterId, Integer brokerId) {
return "CONNECT_JMX_FAILED_BACK_OFF_EVENT_PHY_" + physicalClusterId + "_BROKER_" + brokerId;
}
}

162
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/zookeeper/znode/brokers/BrokerMetadata.java
View File

@@ -1,6 +1,17 @@
package com.xiaojukeji.kafka.manager.common.zookeeper.znode.brokers;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.xiaojukeji.kafka.manager.common.constant.KafkaConstant;
import com.xiaojukeji.kafka.manager.common.entity.ao.common.IpPortData;
import com.xiaojukeji.kafka.manager.common.utils.NumberUtils;
import lombok.Data;
import java.io.Serializable;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* @author zengqiao
@@ -10,7 +21,7 @@ import java.util.List;
* 节点结构:
* {
* "listener_security_protocol_map":{"SASL_PLAINTEXT":"SASL_PLAINTEXT"},
* "endpoints":["SASL_PLAINTEXT://10.179.162.202:9093"],
* "endpoints":["SASL_PLAINTEXT://127.0.0.1:9093"],
* "jmx_port":9999,
* "host":null,
* "timestamp":"1546632983233",
@@ -18,22 +29,48 @@ import java.util.List;
* "version":4,
* "rack": "CY"
* }
*
* {
* "listener_security_protocol_map":{"SASL_PLAINTEXT":"SASL_PLAINTEXT","PLAINTEXT":"PLAINTEXT"},
* "endpoints":["SASL_PLAINTEXT://127.0.0.1:9093","PLAINTEXT://127.0.0.1:9092"],
* "jmx_port":8099,
* "host":"127.0.0.1",
* "timestamp":"1628833925822",
* "port":9092,
* "version":4
* }
*
* {
* "listener_security_protocol_map":{"EXTERNAL":"SASL_PLAINTEXT","INTERNAL":"SASL_PLAINTEXT"},
* "endpoints":["EXTERNAL://127.0.0.1:7092","INTERNAL://127.0.0.1:7093"],
* "jmx_port":8099,
* "host":null,
* "timestamp":"1627289710439",
* "port":-1,
* "version":4
* }
*
*/
public class BrokerMetadata implements Cloneable {
@Data
@JsonIgnoreProperties(ignoreUnknown = true)
public class BrokerMetadata implements Serializable {
private static final long serialVersionUID = 3918113492423375809L;
private long clusterId;
private int brokerId;
private List<String> endpoints;
// <EXTERNAL|INTERNAL, <ip, port>>
private Map<String, IpPortData> endpointMap;
private String host;
private int port;
/*
* ZK上对应的字段就是这个名字, 不要进行修改
*/
private int jmx_port;
@JsonProperty("jmx_port")
private int jmxPort;
private String version;
@@ -41,91 +78,54 @@ public class BrokerMetadata implements Cloneable {
private String rack;
public long getClusterId() {
return clusterId;
@JsonIgnore
public String getExternalHost() {
if (!endpointMap.containsKey(KafkaConstant.EXTERNAL_KEY)) {
return null;
}
return endpointMap.get(KafkaConstant.EXTERNAL_KEY).getIp();
}
public void setClusterId(long clusterId) {
this.clusterId = clusterId;
@JsonIgnore
public String getInternalHost() {
if (!endpointMap.containsKey(KafkaConstant.INTERNAL_KEY)) {
return null;
}
return endpointMap.get(KafkaConstant.INTERNAL_KEY).getIp();
}
public int getBrokerId() {
return brokerId;
}
public static void parseAndUpdateBrokerMetadata(BrokerMetadata brokerMetadata) {
brokerMetadata.setEndpointMap(new HashMap<>());
public void setBrokerId(int brokerId) {
this.brokerId = brokerId;
}
if (brokerMetadata.getEndpoints().isEmpty()) {
return;
}
public List<String> getEndpoints() {
return endpoints;
}
// example EXTERNAL://10.179.162.202:7092
for (String endpoint: brokerMetadata.getEndpoints()) {
int idx1 = endpoint.indexOf("://");
int idx2 = endpoint.lastIndexOf(":");
if (idx1 == -1 || idx2 == -1 || idx1 == idx2) {
continue;
}
public void setEndpoints(List<String> endpoints) {
this.endpoints = endpoints;
}
String brokerHost = endpoint.substring(idx1 + "://".length(), idx2);
String brokerPort = endpoint.substring(idx2 + 1);
public String getHost() {
return host;
}
brokerMetadata.getEndpointMap().put(endpoint.substring(0, idx1), new IpPortData(brokerHost, brokerPort));
public void setHost(String host) {
this.host = host;
}
if (KafkaConstant.EXTERNAL_KEY.equals(endpoint.substring(0, idx1))) {
// 优先使用external的地址进行展示
brokerMetadata.setHost(brokerHost);
brokerMetadata.setPort(NumberUtils.string2Integer(brokerPort));
}
public int getPort() {
return port;
}
public void setPort(int port) {
this.port = port;
}
public int getJmxPort() {
return jmx_port;
}
public void setJmxPort(int jmxPort) {
this.jmx_port = jmxPort;
}
public String getVersion() {
return version;
}
public void setVersion(String version) {
this.version = version;
}
public long getTimestamp() {
return timestamp;
}
public void setTimestamp(long timestamp) {
this.timestamp = timestamp;
}
public String getRack() {
return rack;
}
public void setRack(String rack) {
this.rack = rack;
}
@Override
public String toString() {
return "BrokerMetadata{" +
"clusterId=" + clusterId +
", brokerId=" + brokerId +
", endpoints=" + endpoints +
", host='" + host + '\'' +
", port=" + port +
", jmxPort=" + jmx_port +
", version='" + version + '\'' +
", timestamp=" + timestamp +
", rack='" + rack + '\'' +
'}';
if (null == brokerMetadata.getHost()) {
brokerMetadata.setHost(brokerHost);
brokerMetadata.setPort(NumberUtils.string2Integer(brokerPort));
}
}
}
}