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

Topic基本信息中增加retention.bytes信息

Browse Source
This commit is contained in:
zengqiao
2021-07-02 16:41:57 +08:00
parent eb3b8c4b31
commit 0f746917a7
8 changed files with 119 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -25,6 +25,8 @@ public class TopicCreationConstant {
public static final String TOPIC_RETENTION_TIME_KEY_NAME = "retention.ms";
public static final String TOPIC_RETENTION_BYTES_KEY_NAME = "retention.bytes";
public static final Long DEFAULT_QUOTA = 3 * 1024 * 1024L;
public static Properties createNewProperties(Long retentionTime) {

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

@@ -37,6 +37,8 @@ public class TopicBasicDTO {
private Long retentionTime;
private Long retentionBytes;
public Long getClusterId() {
return clusterId;
}
@@ -157,6 +159,14 @@ public class TopicBasicDTO {
this.retentionTime = retentionTime;
}
public Long getRetentionBytes() {
return retentionBytes;
}
public void setRetentionBytes(Long retentionBytes) {
this.retentionBytes = retentionBytes;
}
@Override
public String toString() {
return "TopicBasicDTO{" +
@@ -166,7 +176,7 @@ public class TopicBasicDTO {
", principals='" + principals + '\'' +
", topicName='" + topicName + '\'' +
", description='" + description + '\'' +
", regionNameList='" + regionNameList + '\'' +
", regionNameList=" + regionNameList +
", score=" + score +
", topicCodeC='" + topicCodeC + '\'' +
", partitionNum=" + partitionNum +
@@ -175,6 +185,7 @@ public class TopicBasicDTO {
", modifyTime=" + modifyTime +
", createTime=" + createTime +
", retentionTime=" + retentionTime +
", retentionBytes=" + retentionBytes +
'}';
}
}

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

@@ -33,6 +33,9 @@ public class TopicBasicVO {
@ApiModelProperty(value = "存储时间(ms)")
private Long retentionTime;
@ApiModelProperty(value = "单分区数据保存大小(Byte)")
private Long retentionBytes;
@ApiModelProperty(value = "创建时间")
private Long createTime;
@@ -62,12 +65,20 @@ public class TopicBasicVO {
this.clusterId = clusterId;
}
public String getTopicCodeC() {
return topicCodeC;
public String getAppId() {
return appId;
}
public void setTopicCodeC(String topicCodeC) {
this.topicCodeC = topicCodeC;
public void setAppId(String appId) {
this.appId = appId;
}
public String getAppName() {
return appName;
}
public void setAppName(String appName) {
this.appName = appName;
}
public Integer getPartitionNum() {
@@ -86,22 +97,6 @@ public class TopicBasicVO {
this.replicaNum = replicaNum;
}
public Long getModifyTime() {
return modifyTime;
}
public void setModifyTime(Long modifyTime) {
this.modifyTime = modifyTime;
}
public Long getCreateTime() {
return createTime;
}
public void setCreateTime(Long createTime) {
this.createTime = createTime;
}
public String getPrincipals() {
return principals;
}
@@ -110,30 +105,6 @@ public class TopicBasicVO {
this.principals = principals;
}
public String getDescription() {
return description;
}
public void setDescription(String description) {
this.description = description;
}
public void setAppId(String appId) {
this.appId = appId;
}
public void setBootstrapServers(String bootstrapServers) {
this.bootstrapServers = bootstrapServers;
}
public String getAppId() {
return appId;
}
public String getBootstrapServers() {
return bootstrapServers;
}
public Long getRetentionTime() {
return retentionTime;
}
@@ -142,12 +113,28 @@ public class TopicBasicVO {
this.retentionTime = retentionTime;
}
public String getAppName() {
return appName;
public Long getRetentionBytes() {
return retentionBytes;
}
public void setAppName(String appName) {
this.appName = appName;
public void setRetentionBytes(Long retentionBytes) {
this.retentionBytes = retentionBytes;
}
public Long getCreateTime() {
return createTime;
}
public void setCreateTime(Long createTime) {
this.createTime = createTime;
}
public Long getModifyTime() {
return modifyTime;
}
public void setModifyTime(Long modifyTime) {
this.modifyTime = modifyTime;
}
public Integer getScore() {
@@ -158,6 +145,30 @@ public class TopicBasicVO {
this.score = score;
}
public String getTopicCodeC() {
return topicCodeC;
}
public void setTopicCodeC(String topicCodeC) {
this.topicCodeC = topicCodeC;
}
public String getDescription() {
return description;
}
public void setDescription(String description) {
this.description = description;
}
public String getBootstrapServers() {
return bootstrapServers;
}
public void setBootstrapServers(String bootstrapServers) {
this.bootstrapServers = bootstrapServers;
}
public List<String> getRegionNameList() {
return regionNameList;
}
@@ -176,6 +187,7 @@ public class TopicBasicVO {
", replicaNum=" + replicaNum +
", principals='" + principals + '\'' +
", retentionTime=" + retentionTime +
", retentionBytes=" + retentionBytes +
", createTime=" + createTime +
", modifyTime=" + modifyTime +
", score=" + score +

43
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/cache/PhysicalClusterMetadataManager.java vendored
View File

@@ -3,10 +3,12 @@ package com.xiaojukeji.kafka.manager.service.cache;
import com.xiaojukeji.kafka.manager.common.bizenum.KafkaBrokerRoleEnum;
import com.xiaojukeji.kafka.manager.common.constant.Constant;
import com.xiaojukeji.kafka.manager.common.constant.KafkaConstant;
import com.xiaojukeji.kafka.manager.common.constant.TopicCreationConstant;
import com.xiaojukeji.kafka.manager.common.entity.KafkaVersion;
import com.xiaojukeji.kafka.manager.common.entity.pojo.ClusterDO;
import com.xiaojukeji.kafka.manager.common.utils.JsonUtils;
import com.xiaojukeji.kafka.manager.common.utils.ListUtils;
import com.xiaojukeji.kafka.manager.common.utils.NumberUtils;
import com.xiaojukeji.kafka.manager.common.utils.ValidateUtils;
import com.xiaojukeji.kafka.manager.common.utils.jmx.JmxConfig;
import com.xiaojukeji.kafka.manager.common.utils.jmx.JmxConnectorWrap;
@@ -56,7 +58,7 @@ public class PhysicalClusterMetadataManager {
private final static Map<Long, Map<String, TopicMetadata>> TOPIC_METADATA_MAP = new ConcurrentHashMap<>();
private final static Map<Long, Map<String, Long>> TOPIC_RETENTION_TIME_MAP = new ConcurrentHashMap<>();
private final static Map<Long, Map<String, Properties>> TOPIC_PROPERTIES_MAP = new ConcurrentHashMap<>();
private final static Map<Long, Map<Integer, BrokerMetadata>> BROKER_METADATA_MAP = new ConcurrentHashMap<>();
@@ -95,7 +97,7 @@ public class PhysicalClusterMetadataManager {
// 初始化topic-map
TOPIC_METADATA_MAP.put(clusterDO.getId(), new ConcurrentHashMap<>());
TOPIC_RETENTION_TIME_MAP.put(clusterDO.getId(), new ConcurrentHashMap<>());
TOPIC_PROPERTIES_MAP.put(clusterDO.getId(), new ConcurrentHashMap<>());
// 初始化cluster-map
CLUSTER_MAP.put(clusterDO.getId(), clusterDO);
@@ -158,7 +160,7 @@ public class PhysicalClusterMetadataManager {
KAFKA_VERSION_MAP.remove(clusterId);
TOPIC_METADATA_MAP.remove(clusterId);
TOPIC_RETENTION_TIME_MAP.remove(clusterId);
TOPIC_PROPERTIES_MAP.remove(clusterId);
CLUSTER_MAP.remove(clusterId);
}
@@ -262,24 +264,45 @@ public class PhysicalClusterMetadataManager {
//---------------------------配置相关元信息--------------
public static void putTopicRetentionTime(Long clusterId, String topicName, Long retentionTime) {
Map<String, Long> timeMap = TOPIC_RETENTION_TIME_MAP.get(clusterId);
if (timeMap == null) {
public static void putTopicProperties(Long clusterId, String topicName, Properties properties) {
if (ValidateUtils.isNull(clusterId) || ValidateUtils.isBlank(topicName) || ValidateUtils.isNull(properties)) {
return;
}
timeMap.put(topicName, retentionTime);
Map<String, Properties> propertiesMap = TOPIC_PROPERTIES_MAP.get(clusterId);
if (ValidateUtils.isNull(propertiesMap)) {
return;
}
propertiesMap.put(topicName, properties);
}
public static Long getTopicRetentionTime(Long clusterId, String topicName) {
Map<String, Long> timeMap = TOPIC_RETENTION_TIME_MAP.get(clusterId);
if (timeMap == null) {
Map<String, Properties> propertiesMap = TOPIC_PROPERTIES_MAP.get(clusterId);
if (ValidateUtils.isNull(propertiesMap)) {
return null;
}
return timeMap.get(topicName);
Properties properties = propertiesMap.get(topicName);
if (ValidateUtils.isNull(properties)) {
return null;
}
return NumberUtils.string2Long(properties.getProperty(TopicCreationConstant.TOPIC_RETENTION_TIME_KEY_NAME));
}
public static Long getTopicRetentionBytes(Long clusterId, String topicName) {
Map<String, Properties> propertiesMap = TOPIC_PROPERTIES_MAP.get(clusterId);
if (ValidateUtils.isNull(propertiesMap)) {
return null;
}
Properties properties = propertiesMap.get(topicName);
if (ValidateUtils.isNull(properties)) {
return null;
}
return NumberUtils.string2Long(properties.getProperty(TopicCreationConstant.TOPIC_RETENTION_BYTES_KEY_NAME));
}
//---------------------------Broker元信息相关--------------

1
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/service/impl/TopicServiceImpl.java
View File

@@ -223,6 +223,7 @@ public class TopicServiceImpl implements TopicService {
basicDTO.setCreateTime(topicMetadata.getCreateTime());
basicDTO.setModifyTime(topicMetadata.getModifyTime());
basicDTO.setRetentionTime(PhysicalClusterMetadataManager.getTopicRetentionTime(clusterId, topicName));
basicDTO.setRetentionBytes(PhysicalClusterMetadataManager.getTopicRetentionBytes(clusterId, topicName));
TopicDO topicDO = topicManagerService.getByTopicName(clusterId, topicName);
if (!ValidateUtils.isNull(topicDO)) {

19
kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/schedule/metadata/FlushTopicRetentionTime.java → kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/schedule/metadata/FlushTopicProperties.java
View File

@@ -14,13 +14,14 @@ import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.util.List;
import java.util.Properties;
/**
* @author zengqiao
* @date 20/7/23
*/
@Component
public class FlushTopicRetentionTime {
public class FlushTopicProperties {
private final static Logger LOGGER = LoggerFactory.getLogger(LogConstant.SCHEDULED_TASK_LOGGER);
@Autowired
@@ -33,7 +34,7 @@ public class FlushTopicRetentionTime {
try {
flush(clusterDO);
} catch (Exception e) {
LOGGER.error("flush topic retention time failed, clusterId:{}.", clusterDO.getId(), e);
LOGGER.error("flush topic properties failed, clusterId:{}.", clusterDO.getId(), e);
}
}
}
@@ -41,22 +42,20 @@ public class FlushTopicRetentionTime {
private void flush(ClusterDO clusterDO) {
ZkConfigImpl zkConfig = PhysicalClusterMetadataManager.getZKConfig(clusterDO.getId());
if (ValidateUtils.isNull(zkConfig)) {
LOGGER.error("flush topic retention time, get zk config failed, clusterId:{}.", clusterDO.getId());
LOGGER.error("flush topic properties, get zk config failed, clusterId:{}.", clusterDO.getId());
return;
}
for (String topicName: PhysicalClusterMetadataManager.getTopicNameList(clusterDO.getId())) {
try {
Long retentionTime = KafkaZookeeperUtils.getTopicRetentionTime(zkConfig, topicName);
if (retentionTime == null) {
LOGGER.warn("get topic retentionTime failed, clusterId:{} topicName:{}.",
clusterDO.getId(), topicName);
Properties properties = KafkaZookeeperUtils.getTopicProperties(zkConfig, topicName);
if (ValidateUtils.isNull(properties)) {
LOGGER.warn("get topic properties failed, clusterId:{} topicName:{}.", clusterDO.getId(), topicName);
continue;
}
PhysicalClusterMetadataManager.putTopicRetentionTime(clusterDO.getId(), topicName, retentionTime);
PhysicalClusterMetadataManager.putTopicProperties(clusterDO.getId(), topicName, properties);
} catch (Exception e) {
LOGGER.error("get topic retentionTime failed, clusterId:{} topicName:{}.",
clusterDO.getId(), topicName, e);
LOGGER.error("get topic properties failed, clusterId:{} topicName:{}.", clusterDO.getId(), topicName, e);
}
}
}

5
kafka-manager-web/src/main/java/com/xiaojukeji/kafka/manager/web/api/versionone/normal/NormalTopicController.java
View File

@@ -61,10 +61,7 @@ public class NormalTopicController {
@ApiOperation(value = "Topic基本信息", notes = "")
@RequestMapping(value = "{clusterId}/topics/{topicName}/basic-info", method = RequestMethod.GET)
@ResponseBody
public Result<TopicBasicVO> getTopicBasic(
@PathVariable Long clusterId,
@PathVariable String topicName,
@RequestParam(value = "isPhysicalClusterId", required = false) Boolean isPhysicalClusterId) {
public Result<TopicBasicVO> getTopicBasic(@PathVariable Long clusterId, @PathVariable String topicName, @RequestParam(value = "isPhysicalClusterId", required = false) Boolean isPhysicalClusterId) {
Long physicalClusterId = logicalClusterMetadataManager.getPhysicalClusterId(clusterId, isPhysicalClusterId);
if (ValidateUtils.isNull(physicalClusterId)) {
return Result.buildFrom(ResultStatus.CLUSTER_NOT_EXIST);

1
kafka-manager-web/src/main/java/com/xiaojukeji/kafka/manager/web/converters/TopicModelConverter.java
View File

@@ -31,6 +31,7 @@ public class TopicModelConverter {
vo.setReplicaNum(dto.getReplicaNum());
vo.setPrincipals(dto.getPrincipals());
vo.setRetentionTime(dto.getRetentionTime());
vo.setRetentionBytes(dto.getRetentionBytes());
vo.setCreateTime(dto.getCreateTime());
vo.setModifyTime(dto.getModifyTime());
vo.setScore(dto.getScore());