support unsubscribe topics while delconsumer in http mode

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

@@ -18,8 +18,10 @@
 package org.apache.eventmesh.runtime.core.protocol.http.consumer;
 
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
 import org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupConf;
 import org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupTopicConf;
@@ -82,4 +84,14 @@ public synchronized void refresh(ConsumerGroupConf consumerGroupConfig) throws E
     public ConsumerGroupConf getConsumerGroupConfig() {
         return consumerGroupConfig;
     }
+
+    public void unsubscribe(String consumerGroup) throws Exception {
+        if(StringUtils.equals(consumerGroupConfig.getConsumerGroup(), consumerGroup)){
+            Set<String> topics = consumerGroupConfig.getConsumerGroupTopicConf().keySet();
+            for (String topic : topics){
+                ConsumerGroupTopicConf consumerGroupTopicConf = consumerGroupConfig.getConsumerGroupTopicConf().get(topic);
+                eventMeshConsumer.unsubscribe(topic, consumerGroupTopicConf.getSubscriptionItem().getMode());
+            }
+        }
+    }
 }

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

@@ -238,10 +238,14 @@ public ConsumerGroupManager getConsumer(String consumerGroup) throws Exception {
      * @param consumerGroup
      */
     public synchronized void delConsumer(String consumerGroup) throws Exception {
+        logger.info("start delConsumer with consumerGroup {}", consumerGroup);
         if(consumerTable.containsKey(consumerGroup)) {
             ConsumerGroupManager cgm = consumerTable.remove(consumerGroup);
+            logger.info("start unsubscribe topic with consumer group manager {}", JSONObject.toJSONString(cgm));
+            cgm.unsubscribe(consumerGroup);
             cgm.shutdown();
         }
+        logger.info("end delConsumer with consumerGroup {}", consumerGroup);
     }
 
     @Subscribe