wz2cool commented on issue #3660:
URL: https://github.com/apache/rocketmq/issues/3660#issuecomment-996665021
i write a example to filter valid consumer group
```
private Set<String> listValidConsumerGroups(String topic)
throws RemotingException, MQClientException,
InterruptedException, MQBrokerException {
final TopicRouteData topicRouteData =
this.mqAdmin.examineTopicRouteInfo(topic);
if (CollectionUtils.isEmpty(topicRouteData.getBrokerDatas())) {
return new HashSet<>();
}
Set<String> validGroupSet = new HashSet<>();
for (BrokerData brokerData : topicRouteData.getBrokerDatas()) {
for (Map.Entry<Long, String> entry :
brokerData.getBrokerAddrs().entrySet()) {
String brokerAddr = entry.getValue();
long timeoutMillis = TimeUnit.SECONDS.toMillis(1);
final SubscriptionGroupWrapper allSubscriptionGroup =
this.mqAdmin.getAllSubscriptionGroup(brokerAddr, timeoutMillis);
ConcurrentMap<String, SubscriptionGroupConfig>
subscriptionGroupTable = allSubscriptionGroup.getSubscriptionGroupTable();
validGroupSet.addAll(subscriptionGroupTable.keySet());
}
}
final GroupList consumeGroupList =
this.mqAdmin.queryTopicConsumeByWho(topic);
// 删除消费组是无法删除topic 和消费组的关系的， 所以我们要和现在的组做一次交集运算
final HashSet<String> relationGroups =
consumeGroupList.getGroupList();
validGroupSet.retainAll(relationGroups);
return validGroupSet;
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
