[Bug #590] Consumer subscription topic is invalid (#590) (#592)

* [Bug #590] Consumer subscription topic is invalid (#590) * [Bug #590] Consumer subscription topic is invalid (#590) close #590
apache · Nov 16, 2021 · 46f973d · 46f973d
1 parent ceaf33b
commit 46f973d
Showing 1 changed file with 11 additions and 2 deletions.
diff --git a/...in/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SubscribeProcessor.java b/...in/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SubscribeProcessor.java
@@ -192,16 +192,25 @@ public void processRequest(ChannelHandlerContext ctx, AsyncContext<HttpCommand>
  for (String key : map.keySet()) {
  if (StringUtils.equals(subTopic.getTopic(), key)) {
  ConsumerGroupTopicConf latestTopicConf = new ConsumerGroupTopicConf();
- ConsumerGroupTopicConf currentTopicConf = map.get(key);
  latestTopicConf.setConsumerGroup(consumerGroup);
  latestTopicConf.setTopic(subTopic.getTopic());
  latestTopicConf.setSubscriptionItem(subTopic);
  latestTopicConf.setUrls(new HashSet<>(Arrays.asList(url)));
- latestTopicConf.getUrls().addAll(currentTopicConf.getUrls());
 
+ ConsumerGroupTopicConf currentTopicConf = map.get(key);
+ latestTopicConf.getUrls().addAll(currentTopicConf.getUrls());
  latestTopicConf.setIdcUrls(idcUrls);
 
  map.put(key, latestTopicConf);
+ } else {
+ //If there are multiple topics, append it
+ ConsumerGroupTopicConf newTopicConf = new ConsumerGroupTopicConf();
+ newTopicConf.setConsumerGroup(consumerGroup);
+ newTopicConf.setTopic(subTopic.getTopic());
+ newTopicConf.setSubscriptionItem(subTopic);
+ newTopicConf.setUrls(new HashSet<>(Arrays.asList(url)));
+ newTopicConf.setIdcUrls(idcUrls);
+ map.put(subTopic.getTopic(), newTopicConf);
  }
  }
  }