mirror of
https://github.com/didi/KnowStreaming.git
synced 2025-12-24 11:52:08 +08:00
fix NPE when flush logical cluster and physical cluster not in cache or not exist
This commit is contained in:
@@ -208,7 +208,8 @@ public class LogicalClusterMetadataManager {
|
|||||||
// 计算逻辑集群到Topic名称的映射
|
// 计算逻辑集群到Topic名称的映射
|
||||||
Set<String> topicNameSet = PhysicalClusterMetadataManager.getBrokerTopicNum(
|
Set<String> topicNameSet = PhysicalClusterMetadataManager.getBrokerTopicNum(
|
||||||
logicalClusterDO.getClusterId(),
|
logicalClusterDO.getClusterId(),
|
||||||
brokerIdSet);
|
brokerIdSet
|
||||||
|
);
|
||||||
LOGICAL_CLUSTER_ID_TOPIC_NAME_MAP.put(logicalClusterDO.getId(), topicNameSet);
|
LOGICAL_CLUSTER_ID_TOPIC_NAME_MAP.put(logicalClusterDO.getId(), topicNameSet);
|
||||||
|
|
||||||
// 计算Topic名称到逻辑集群的映射
|
// 计算Topic名称到逻辑集群的映射
|
||||||
|
|||||||
@@ -539,9 +539,12 @@ public class PhysicalClusterMetadataManager {
|
|||||||
}
|
}
|
||||||
|
|
||||||
public static Set<String> getBrokerTopicNum(Long clusterId, Set<Integer> brokerIdSet) {
|
public static Set<String> getBrokerTopicNum(Long clusterId, Set<Integer> brokerIdSet) {
|
||||||
Set<String> topicNameSet = new HashSet<>();
|
|
||||||
|
|
||||||
Map<String, TopicMetadata> metadataMap = TOPIC_METADATA_MAP.get(clusterId);
|
Map<String, TopicMetadata> metadataMap = TOPIC_METADATA_MAP.get(clusterId);
|
||||||
|
if (metadataMap == null) {
|
||||||
|
return new HashSet<>();
|
||||||
|
}
|
||||||
|
|
||||||
|
Set<String> topicNameSet = new HashSet<>();
|
||||||
for (String topicName: metadataMap.keySet()) {
|
for (String topicName: metadataMap.keySet()) {
|
||||||
try {
|
try {
|
||||||
TopicMetadata tm = metadataMap.get(topicName);
|
TopicMetadata tm = metadataMap.get(topicName);
|
||||||
|
|||||||
Reference in New Issue
Block a user