This is an automated email from the ASF dual-hosted git repository.
mikexue pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/incubator-eventmesh.git
The following commit(s) were added to refs/heads/develop by this push:
new 46f973d [Bug #590] Consumer subscription topic is invalid (#590)
(#592)
46f973d is described below
commit 46f973d30f308640a4d024e6da4fada2c505d691
Author: hagsyn <[email protected]>
AuthorDate: Tue Nov 16 15:28:55 2021 +0800
[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
---
.../core/protocol/http/processor/SubscribeProcessor.java | 13 +++++++++++--
1 file changed, 11 insertions(+), 2 deletions(-)
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SubscribeProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SubscribeProcessor.java
index 4be4fb2..d91b462 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SubscribeProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SubscribeProcessor.java
@@ -192,16 +192,25 @@ public class SubscribeProcessor implements
HttpRequestProcessor {
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);
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
