[Bug apache#590] Consumer subscription topic is invalid (apache#590) (a…

…pache#592) * [Bug apache#590] Consumer subscription topic is invalid (apache#590) * [Bug apache#590] Consumer subscription topic is invalid (apache#590) close apache#590
xwm1992 · Dec 27, 2021 · b5a2ccf · b5a2ccf
1 parent c9c4982
commit b5a2ccf
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);
                         }
                     }
                 }