[Issue #768] fix issue in update HTTP subscriber

eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/SubscriptionItem.java

@@ -17,7 +17,7 @@
 
 package org.apache.eventmesh.common.protocol;
 
-public class SubscriptionItem {
+public class SubscriptionItem implements Cloneable {
 
     private String topic;
 
@@ -66,6 +66,11 @@ public String toString() {
                 + ", type=" + type
                 + '}';
     }
+
+    @Override
+    public SubscriptionItem clone() {
+        return new SubscriptionItem(topic, mode, type);
+    }
 }
 
 

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/consumergroup/ConsumerGroupConf.java

@@ -19,10 +19,11 @@
 
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
 
 import com.google.common.collect.Maps;
 
-public class ConsumerGroupConf {
+public class ConsumerGroupConf implements Cloneable {
     //eg . 5013-1A0
     private String consumerGroup;
 
@@ -76,4 +77,15 @@ public String toString() {
                 .append(",consumerGroupTopicConf=").append(consumerGroupTopicConf).append("}");
         return sb.toString();
     }
+
+    @Override
+    public ConsumerGroupConf clone() {
+        ConsumerGroupConf clone = new ConsumerGroupConf(consumerGroup);
+
+        Map<String, ConsumerGroupTopicConf> cloneGroupTopicConf = new ConcurrentHashMap<>();
+        consumerGroupTopicConf.forEach((topic, groupTopicConf) -> cloneGroupTopicConf.put(topic, groupTopicConf.clone()));
+
+        clone.setConsumerGroupTopicConf(cloneGroupTopicConf);
+        return clone;
+    }
 }

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/consumergroup/ConsumerGroupTopicConf.java

@@ -19,6 +19,7 @@
 
 import org.apache.eventmesh.common.protocol.SubscriptionItem;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -30,7 +31,7 @@
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
 
-public class ConsumerGroupTopicConf {
+public class ConsumerGroupTopicConf implements Cloneable {
 
     public static Logger logger = LoggerFactory.getLogger(ConsumerGroupTopicConf.class);
 
@@ -126,4 +127,19 @@ public Set<String> getUrls() {
     public void setUrls(Set<String> urls) {
         this.urls = urls;
     }
+
+    @Override
+    public ConsumerGroupTopicConf clone() {
+        ConsumerGroupTopicConf clone = new ConsumerGroupTopicConf();
+        clone.setConsumerGroup(consumerGroup);
+        clone.setTopic(topic);
+        clone.setSubscriptionItem(subscriptionItem.clone());
+
+        Map<String, List<String>> cloneIdcUrls = Maps.newConcurrentMap();
+        idcUrls.forEach((idc, urls) -> cloneIdcUrls.put(idc, new ArrayList<>(urls)));
+        clone.setIdcUrls(cloneIdcUrls);
+
+        clone.setUrls(Sets.newConcurrentHashSet(urls));
+        return clone;
+    }
 }

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/consumer/ConsumerManager.java

@@ -192,7 +192,7 @@ public void notifyConsumerManager(String consumerGroup,
             ConsumerGroupStateEvent notification = new ConsumerGroupStateEvent();
             notification.action = ConsumerGroupStateEvent.ConsumerGroupStateAction.NEW;
             notification.consumerGroup = consumerGroup;
-            notification.consumerGroupConfig = latestConsumerGroupConfig;
+            notification.consumerGroupConfig = latestConsumerGroupConfig.clone();
             eventMeshHTTPServer.getEventBus().post(notification);
             return;
         }
@@ -201,7 +201,7 @@ public void notifyConsumerManager(String consumerGroup,
             ConsumerGroupStateEvent notification = new ConsumerGroupStateEvent();
             notification.action = ConsumerGroupStateEvent.ConsumerGroupStateAction.CHANGE;
             notification.consumerGroup = consumerGroup;
-            notification.consumerGroupConfig = latestConsumerGroupConfig;
+            notification.consumerGroupConfig = latestConsumerGroupConfig.clone();
             eventMeshHTTPServer.getEventBus().post(notification);
             return;
         }