resolve conflict

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/consumer/SubscriptionManager.java

@@ -36,6 +36,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.locks.ReentrantLock;
 
 import lombok.extern.slf4j.Slf4j;
 
@@ -56,6 +57,8 @@ public class SubscriptionManager {
      */
     private final ConcurrentHashMap<String, List<Client>> localClientInfoMapping = new ConcurrentHashMap<>(64);
 
+    private final ReentrantLock lock = new ReentrantLock();
+
     public SubscriptionManager(boolean isEventMeshServerMetaStorageEnable, MetaStorage metaStorage) {
         this.isEventMeshServerMetaStorageEnable = isEventMeshServerMetaStorageEnable;
         this.metaStorage = metaStorage;
@@ -71,82 +74,87 @@ public ConcurrentHashMap<String, List<Client>> getLocalClientInfoMapping() {
 
     public void registerClient(final ClientInfo clientInfo, final String consumerGroup,
         final List<SubscriptionItem> subscriptionItems, final String url) {
-        for (final SubscriptionItem subscription : subscriptionItems) {
-            final String groupTopicKey = consumerGroup + "@" + subscription.getTopic();
-
-            List<Client> localClients = localClientInfoMapping.get(groupTopicKey);
-
-            if (localClients == null) {
-                localClientInfoMapping.putIfAbsent(groupTopicKey, new ArrayList<>());
-                localClients = localClientInfoMapping.get(groupTopicKey);
-            }
-
-            boolean isContains = false;
-            for (final Client localClient : localClients) {
-                // TODO: compare the whole Client would be better?
-                if (StringUtils.equals(localClient.getUrl(), url)) {
-                    isContains = true;
-                    localClient.setLastUpTime(new Date());
-                    break;
+        lock.lock();
+        try {
+            for (final SubscriptionItem subscription : subscriptionItems) {
+                final String groupTopicKey = consumerGroup + "@" + subscription.getTopic();
+                List<Client> localClients = localClientInfoMapping.computeIfAbsent(groupTopicKey, (groupTopicKeyInner) ->
+                    new ArrayList<>()
+                );
+                boolean isContains = false;
+                for (final Client localClient : localClients) {
+                    // TODO: compare the whole Client would be better?
+                    if (StringUtils.equals(localClient.getUrl(), url)) {
+                        isContains = true;
+                        localClient.setLastUpTime(new Date());
+                        break;
+                    }
                 }
-            }
 
-            if (!isContains) {
-                Client client = new Client();
-                client.setEnv(clientInfo.getEnv());
-                client.setIdc(clientInfo.getIdc());
-                client.setSys(clientInfo.getSys());
-                client.setIp(clientInfo.getIp());
-                client.setPid(clientInfo.getPid());
-                client.setConsumerGroup(consumerGroup);
-                client.setTopic(subscription.getTopic());
-                client.setUrl(url);
-                client.setLastUpTime(new Date());
-                localClients.add(client);
+                if (!isContains) {
+                    Client client = new Client();
+                    client.setEnv(clientInfo.getEnv());
+                    client.setIdc(clientInfo.getIdc());
+                    client.setSys(clientInfo.getSys());
+                    client.setIp(clientInfo.getIp());
+                    client.setPid(clientInfo.getPid());
+                    client.setConsumerGroup(consumerGroup);
+                    client.setTopic(subscription.getTopic());
+                    client.setUrl(url);
+                    client.setLastUpTime(new Date());
+                    localClients.add(client);
+                }
             }
+        } finally {
+            lock.unlock();
         }
     }
 
     public void updateSubscription(ClientInfo clientInfo, String consumerGroup,
         String url, List<SubscriptionItem> subscriptionList) {
-        for (final SubscriptionItem subscription : subscriptionList) {
-            final List<Client> groupTopicClients = localClientInfoMapping
-                .get(consumerGroup + "@" + subscription.getTopic());
+        lock.lock();
+        try {
+            for (final SubscriptionItem subscription : subscriptionList) {
+                final List<Client> groupTopicClients = localClientInfoMapping
+                    .get(consumerGroup + "@" + subscription.getTopic());
 
-            if (CollectionUtils.isEmpty(groupTopicClients)) {
-                log.error("group {} topic {} clients is empty", consumerGroup, subscription);
-            }
+                if (CollectionUtils.isEmpty(groupTopicClients)) {
+                    log.error("group {} topic {} clients is empty", consumerGroup, subscription);
+                }
 
-            ConsumerGroupConf consumerGroupConf = localConsumerGroupMapping.get(consumerGroup);
-            if (consumerGroupConf == null) {
-                // new subscription
-                ConsumerGroupConf prev = localConsumerGroupMapping.putIfAbsent(consumerGroup, new ConsumerGroupConf(consumerGroup));
-                if (prev == null) {
-                    log.info("add new subscription, consumer group: {}", consumerGroup);
+                ConsumerGroupConf consumerGroupConf = localConsumerGroupMapping.get(consumerGroup);
+                if (consumerGroupConf == null) {
+                    // new subscription
+                    ConsumerGroupConf prev = localConsumerGroupMapping.putIfAbsent(consumerGroup, new ConsumerGroupConf(consumerGroup));
+                    if (prev == null) {
+                        log.info("add new subscription, consumer group: {}", consumerGroup);
+                    }
+                    consumerGroupConf = localConsumerGroupMapping.get(consumerGroup);
                 }
-                consumerGroupConf = localConsumerGroupMapping.get(consumerGroup);
-            }
 
-            ConsumerGroupTopicConf consumerGroupTopicConf = consumerGroupConf.getConsumerGroupTopicConf()
-                .get(subscription.getTopic());
-            if (consumerGroupTopicConf == null) {
-                consumerGroupConf.getConsumerGroupTopicConf().computeIfAbsent(subscription.getTopic(), (topic) -> {
-                    ConsumerGroupTopicConf newTopicConf = new ConsumerGroupTopicConf();
-                    newTopicConf.setConsumerGroup(consumerGroup);
-                    newTopicConf.setTopic(topic);
-                    newTopicConf.setSubscriptionItem(subscription);
-                    log.info("add new {}", newTopicConf);
-                    return newTopicConf;
-                });
-                consumerGroupTopicConf = consumerGroupConf.getConsumerGroupTopicConf().get(subscription.getTopic());
-            }
+                ConsumerGroupTopicConf consumerGroupTopicConf = consumerGroupConf.getConsumerGroupTopicConf()
+                    .get(subscription.getTopic());
+                if (consumerGroupTopicConf == null) {
+                    consumerGroupConf.getConsumerGroupTopicConf().computeIfAbsent(subscription.getTopic(), (topic) -> {
+                        ConsumerGroupTopicConf newTopicConf = new ConsumerGroupTopicConf();
+                        newTopicConf.setConsumerGroup(consumerGroup);
+                        newTopicConf.setTopic(topic);
+                        newTopicConf.setSubscriptionItem(subscription);
+                        log.info("add new {}", newTopicConf);
+                        return newTopicConf;
+                    });
+                    consumerGroupTopicConf = consumerGroupConf.getConsumerGroupTopicConf().get(subscription.getTopic());
+                }
 
-            consumerGroupTopicConf.getUrls().add(url);
-            if (!consumerGroupTopicConf.getIdcUrls().containsKey(clientInfo.getIdc())) {
-                consumerGroupTopicConf.getIdcUrls().putIfAbsent(clientInfo.getIdc(), new ArrayList<>());
+                consumerGroupTopicConf.getUrls().add(url);
+                if (!consumerGroupTopicConf.getIdcUrls().containsKey(clientInfo.getIdc())) {
+                    consumerGroupTopicConf.getIdcUrls().putIfAbsent(clientInfo.getIdc(), new ArrayList<>());
+                }
+                // TODO: idcUrl list is not thread-safe
+                consumerGroupTopicConf.getIdcUrls().get(clientInfo.getIdc()).add(url);
             }
-            // TODO: idcUrl list is not thread-safe
-            consumerGroupTopicConf.getIdcUrls().get(clientInfo.getIdc()).add(url);
+        } finally {
+            lock.unlock();
         }
     }
 

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalSubscribeEventProcessor.java

@@ -155,33 +155,32 @@ public void handler(final HandlerService.HandlerSpecific handlerSpecific, final
             return;
         }
 
-        synchronized (eventMeshHTTPServer.getSubscriptionManager().getLocalClientInfoMapping()) {
-            ClientInfo clientInfo = getClientInfo(requestWrapper);
-            SubscriptionManager subscriptionManager = eventMeshHTTPServer.getSubscriptionManager();
-            subscriptionManager.registerClient(clientInfo, consumerGroup, subscriptionList, url);
-            subscriptionManager.updateSubscription(clientInfo, consumerGroup, url, subscriptionList);
-
-            final long startTime = System.currentTimeMillis();
-            try {
-                // subscription relationship change notification
-                eventMeshHTTPServer.getConsumerManager().notifyConsumerManager(consumerGroup,
-                    eventMeshHTTPServer.getSubscriptionManager().getLocalConsumerGroupMapping().get(consumerGroup));
-                responseBodyMap.put(EventMeshConstants.RET_CODE, EventMeshRetCode.SUCCESS.getRetCode());
-                responseBodyMap.put(EventMeshConstants.RET_MSG, EventMeshRetCode.SUCCESS.getErrMsg());
-
-                handlerSpecific.sendResponse(responseHeaderMap, responseBodyMap);
-
-            } catch (Exception e) {
-                log.error("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|url={}",
-                    System.currentTimeMillis() - startTime, JsonUtils.toJSONString(subscriptionList), url, e);
-
-                handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_SUBSCRIBE_ERR, responseHeaderMap, responseBodyMap, null);
-            }
+        ClientInfo clientInfo = getClientInfo(requestWrapper);
+        SubscriptionManager subscriptionManager = eventMeshHTTPServer.getSubscriptionManager();
+        subscriptionManager.registerClient(clientInfo, consumerGroup, subscriptionList, url);
+        subscriptionManager.updateSubscription(clientInfo, consumerGroup, url, subscriptionList);
+
+        final long startTime = System.currentTimeMillis();
+        try {
+            // subscription relationship change notification
+            eventMeshHTTPServer.getConsumerManager().notifyConsumerManager(consumerGroup,
+                eventMeshHTTPServer.getSubscriptionManager().getLocalConsumerGroupMapping().get(consumerGroup));
+            responseBodyMap.put(EventMeshConstants.RET_CODE, EventMeshRetCode.SUCCESS.getRetCode());
+            responseBodyMap.put(EventMeshConstants.RET_MSG, EventMeshRetCode.SUCCESS.getErrMsg());
 
-            // Update service metadata
-            eventMeshHTTPServer.getSubscriptionManager().updateMetaData();
+            handlerSpecific.sendResponse(responseHeaderMap, responseBodyMap);
+
+        } catch (Exception e) {
+            log.error("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|url={}",
+                System.currentTimeMillis() - startTime, JsonUtils.toJSONString(subscriptionList), url, e);
+
+            handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_SUBSCRIBE_ERR, responseHeaderMap, responseBodyMap, null);
         }
 
+        // Update service metadata
+        eventMeshHTTPServer.getSubscriptionManager().updateMetaData();
+
+
     }
 
     @Override
@@ -192,6 +191,7 @@ public String[] paths() {
     @Override
     public Executor executor() {
         return eventMeshHTTPServer.getHttpThreadPoolGroup().getClientManageExecutor();
+
     }
 
     private ClientInfo getClientInfo(final HttpEventWrapper requestWrapper) {

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SubscribeProcessor.java

@@ -155,46 +155,44 @@ public void processRequest(final ChannelHandlerContext ctx, final AsyncContext<H
         }
 
         SubscriptionManager subscriptionManager = eventMeshHTTPServer.getSubscriptionManager();
-        synchronized (subscriptionManager.getLocalClientInfoMapping()) {
-            ClientInfo clientInfo = getClientInfo(subscribeRequestHeader);
-            subscriptionManager.registerClient(clientInfo, consumerGroup, subTopicList, url);
-            subscriptionManager.updateSubscription(clientInfo, consumerGroup, url, subTopicList);
-
-            final long startTime = System.currentTimeMillis();
-            HttpSummaryMetrics summaryMetrics = eventMeshHTTPServer.getMetrics().getSummaryMetrics();
-            try {
-                // subscription relationship change notification
-                eventMeshHTTPServer.getConsumerManager().notifyConsumerManager(consumerGroup,
-                    subscriptionManager.getLocalConsumerGroupMapping().get(consumerGroup));
-
-                final CompleteHandler<HttpCommand> handler = httpCommand -> {
-                    try {
-                        log.debug("{}", httpCommand);
-                        eventMeshHTTPServer.sendResponse(ctx, httpCommand.httpResponse());
-
-                        summaryMetrics.recordHTTPReqResTimeCost(System.currentTimeMillis() - request.getReqTime());
-                    } catch (Exception ex) {
-                        log.error("onResponse error", ex);
-                    }
-                };
-
-                responseEventMeshCommand = request.createHttpCommandResponse(EventMeshRetCode.SUCCESS);
-                asyncContext.onComplete(responseEventMeshCommand, handler);
-            } catch (Exception e) {
-                completeResponse(request, asyncContext, subscribeResponseHeader,
-                    EventMeshRetCode.EVENTMESH_SUBSCRIBE_ERR,
-                    EventMeshRetCode.EVENTMESH_SUBSCRIBE_ERR.getErrMsg() + EventMeshUtil.stackTrace(e, 2),
-                    SubscribeResponseBody.class);
-                final long endTime = System.currentTimeMillis();
-
-                log.error("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
-                    endTime - startTime, JsonUtils.toJSONString(subscribeRequestBody.getTopics()), subscribeRequestBody.getUrl(), e);
+        ClientInfo clientInfo = getClientInfo(subscribeRequestHeader);
+        subscriptionManager.registerClient(clientInfo, consumerGroup, subTopicList, url);
+        subscriptionManager.updateSubscription(clientInfo, consumerGroup, url, subTopicList);
+
+        final long startTime = System.currentTimeMillis();
+        HttpSummaryMetrics summaryMetrics = eventMeshHTTPServer.getMetrics().getSummaryMetrics();
+        try {
+            // subscription relationship change notification
+            eventMeshHTTPServer.getConsumerManager().notifyConsumerManager(consumerGroup,
+                subscriptionManager.getLocalConsumerGroupMapping().get(consumerGroup));
+
+            final CompleteHandler<HttpCommand> handler = httpCommand -> {
+                try {
+                    log.debug("{}", httpCommand);
+                    eventMeshHTTPServer.sendResponse(ctx, httpCommand.httpResponse());
+
+                    summaryMetrics.recordHTTPReqResTimeCost(System.currentTimeMillis() - request.getReqTime());
+                } catch (Exception ex) {
+                    log.error("onResponse error", ex);
+                }
+            };
+
+            responseEventMeshCommand = request.createHttpCommandResponse(EventMeshRetCode.SUCCESS);
+            asyncContext.onComplete(responseEventMeshCommand, handler);
+        } catch (Exception e) {
+            completeResponse(request, asyncContext, subscribeResponseHeader,
+                EventMeshRetCode.EVENTMESH_SUBSCRIBE_ERR,
+                EventMeshRetCode.EVENTMESH_SUBSCRIBE_ERR.getErrMsg() + EventMeshUtil.stackTrace(e, 2),
+                SubscribeResponseBody.class);
+            final long endTime = System.currentTimeMillis();
+            log.error("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
+                endTime - startTime, JsonUtils.toJSONString(subscribeRequestBody.getTopics()), subscribeRequestBody.getUrl(), e);
+            summaryMetrics.recordSendMsgFailed();
+            summaryMetrics.recordSendMsgCost(endTime - startTime);
 
-                summaryMetrics.recordSendMsgFailed();
-                summaryMetrics.recordSendMsgCost(endTime - startTime);
-            }
-            eventMeshHTTPServer.getSubscriptionManager().updateMetaData();
         }
+        eventMeshHTTPServer.getSubscriptionManager().updateMetaData();
+
     }
 
     @Override