support unsubscribe topics while delconsumer in http mode (apache#396)

* [ISSUE apache#325]Update gradle configuration for publishing package to maven repository * update build.gradle * update build.gradle and gradle.properties * update build.gradle and gradle.properties for publish to maven repository * * update gradle version for instructions * fix: dist task exception * [ISSUE apache#329]Missing Log4j dependency * update eventmesh-runtime.png * support unsubscribe topics while delconsumer in http mode
jjz921024 · Jul 25, 2021 · 741cdfb · 741cdfb
1 parent 8eaa52b
commit 741cdfb
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 0 deletions.
diff --git a/...n/java/org/apache/eventmesh/runtime/core/protocol/http/consumer/ConsumerGroupManager.java b/...n/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());
+            }
+        }
+    }
 }
diff --git a/...c/main/java/org/apache/eventmesh/runtime/core/protocol/http/consumer/ConsumerManager.java b/...c/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