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

[ISSUE-624]过滤掉不存在的Topic(#624)

Browse Source 
同步Group元信息时，如果Topic已经不存在了，则过滤掉该Group+Topic信息
This commit is contained in:
zengqiao
2022-09-28 10:39:33 +08:00
parent 0f07ddedaf
commit 9d33c725ad
1 changed files with 23 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncKafkaGroupTask.java
View File

@@ -12,10 +12,10 @@ import com.xiaojukeji.know.streaming.km.common.exception.AdminOperateException;
import com.xiaojukeji.know.streaming.km.common.exception.NotExistException;
import com.xiaojukeji.know.streaming.km.common.utils.ValidateUtils;
import com.xiaojukeji.know.streaming.km.core.service.group.GroupService;
import com.xiaojukeji.know.streaming.km.core.service.topic.TopicService;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.TopicPartition;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.transaction.annotation.Transactional;
import java.util.*;
import java.util.stream.Collectors;
@@ -33,6 +33,9 @@ public class SyncKafkaGroupTask extends AbstractAsyncMetadataDispatchTask {
@Autowired
private GroupService groupService;
@Autowired
private TopicService topicService;
@Override
public TaskResult processClusterTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) throws Exception {
@@ -53,6 +56,7 @@ public class SyncKafkaGroupTask extends AbstractAsyncMetadataDispatchTask {
private TaskResult updateGroupMembersTask(ClusterPhy clusterPhy, List<String> groupNameList, long triggerTimeUnitMs) {
List<GroupMemberPO> groupMemberPOList = new ArrayList<>();
TaskResult tr = TaskResult.SUCCESS;
for (String groupName : groupNameList) {
try {
List<GroupMemberPO> poList = this.getGroupMembers(clusterPhy.getId(), groupName, new Date(triggerTimeUnitMs));
@@ -62,7 +66,10 @@ public class SyncKafkaGroupTask extends AbstractAsyncMetadataDispatchTask {
tr = TaskResult.FAIL;
}
}
groupMemberPOList = this.filterGroupIfTopicNotExist(clusterPhy.getId(), groupMemberPOList);
groupService.batchReplace(groupMemberPOList);
return tr;
}
@@ -112,4 +119,17 @@ public class SyncKafkaGroupTask extends AbstractAsyncMetadataDispatchTask {
return new ArrayList<>(groupMap.values());
}
private List<GroupMemberPO> filterGroupIfTopicNotExist(Long clusterPhyId, List<GroupMemberPO> poList) {
if (poList.isEmpty()) {
return poList;
}
// 集群Topic集合
Set<String> dbTopicSet = topicService.listTopicsFromDB(clusterPhyId).stream().map(elem -> elem.getTopicName()).collect(Collectors.toSet());
dbTopicSet.add(""); //兼容没有消费Topic的group
// 过滤Topic不存在的消费组
return poList.stream().filter(elem -> dbTopicSet.contains(elem.getTopicName())).collect(Collectors.toList());
}
}