Repository: apache/eventmesh
Issue: [Bug] Possible null pointer dereference

### Search before asking

- [X] I had searched in the [issues](https://github.com/apache/eventmesh/issues?q=is%3Aissue) and found no similar issues.


### Environment

Mac

### EventMesh version

master

### What happened

The return value from a method is dereferenced without a null check, and the return value of that method is one that should generally be checked for null. This may lead to a NullPointerException when the code is executed.

The bug appears in the package org.apache.eventmesh.runtime.core.protocol.http.processor many times.

### How to reproduce

It will happen if the return value is null.

### Debug logs

_No response_

### Are you willing to submit PR?

- [X] Yes I am willing to submit a PR!

Changed files:
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HandlerService.java
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/RemoteUnSubscribeEventProcessor.java
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncRemoteEventProcessor.java
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalUnSubscribeEventProcessor.java
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/inf/AbstractEventProcessor.java
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ReplyMessageProcessor.java
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalSubscribeEventProcessor.java
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageV2Processor.java
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/RemoteSubscribeEventProcessor.java

diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageV2Processor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageV2Processor.java
index a9935323b..482a4afc5 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageV2Processor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageV2Processor.java
@@ -63,7 +63,7 @@ public class BatchSendMessageV2Processor implements HttpRequestProcessor {
 
     public Logger aclLogger = LoggerFactory.getLogger("acl");
 
-    private EventMeshHTTPServer eventMeshHTTPServer;
+    private final EventMeshHTTPServer eventMeshHTTPServer;
 
     public BatchSendMessageV2Processor(EventMeshHTTPServer eventMeshHTTPServer) {
         this.eventMeshHTTPServer = eventMeshHTTPServer;
@@ -175,12 +175,9 @@ public class BatchSendMessageV2Processor implements HttpRequestProcessor {
         //do acl check
         if (eventMeshHTTPServer.getEventMeshHttpConfiguration().isEventMeshServerSecurityEnable()) {
             String remoteAddr = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
-            String user = event.getExtension(ProtocolKey.ClientInstanceKey.USERNAME) == null ? "" :
-                event.getExtension(ProtocolKey.ClientInstanceKey.USERNAME).toString();
-            String pass = event.getExtension(ProtocolKey.ClientInstanceKey.PASSWD) == null ? "" :
-                event.getExtension(ProtocolKey.ClientInstanceKey.PASSWD).toString();
-            String subsystem = event.getExtension(ProtocolKey.ClientInstanceKey.SYS) == null ? "" :
-                event.getExtension(ProtocolKey.ClientInstanceKey.SYS).toString();
+            String user = getExtension(event, ProtocolKey.ClientInstanceKey.USERNAME);
+            String pass = getExtension(event, ProtocolKey.ClientInstanceKey.PASSWD);
+            String subsystem = getExtension(event, ProtocolKey.ClientInstanceKey.SYS);
             try {
                 Acl.doAclCheckInHttpSend(remoteAddr, user, pass, subsystem, topic, requestCode);
             } catch (Exception e) {
@@ -226,7 +223,7 @@ public class BatchSendMessageV2Processor implements HttpRequestProcessor {
 
         String defaultTTL = String.valueOf(EventMeshConstants.DEFAULT_MSG_TTL_MILLS);
         // todo: use hashmap to avoid copy
-        String ttlValue = event.getExtension(SendMessageRequestBody.TTL) == null ? "" : event.getExtension(SendMessageRequestBody.TTL).toString();
+        String ttlValue = getExtension(event, SendMessageRequestBody.TTL);
         if (StringUtils.isBlank(ttlValue) && !StringUtils.isNumeric(ttlValue)) {
             event = CloudEventBuilder.from(event).withExtension(SendMessageRequestBody.TTL, defaultTTL)
                 .build();
@@ -309,6 +306,11 @@ public class BatchSendMessageV2Processor implements HttpRequestProcessor {
 
     }
 
+    private String getExtension(CloudEvent event, String protocolKey) {
+        Object extension = event.getExtension(protocolKey);
+        return Objects.isNull(extension) ? "" : extension.toString();
+    }
+
     @Override
     public boolean rejectRequest() {
         return false;
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HandlerService.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HandlerService.java
index f69869674..75dc3aaa5 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HandlerService.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HandlerService.java
@@ -38,6 +38,7 @@ import java.util.HashMap;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ThreadPoolExecutor;
 
@@ -187,8 +188,12 @@ public class HandlerService {
                 if (length > 0) {
                     byte[] body = new byte[length];
                     fullHttpRequest.content().readBytes(body);
-                    JsonUtils.deserialize(new String(body, Constants.DEFAULT_CHARSET), new TypeReference<Map<String, Object>>() {
-                    }).forEach(bodyMap::put);
+                    Optional
+                        .ofNullable(JsonUtils.deserialize(
+                            new String(body, Constants.DEFAULT_CHARSET),
+                            new TypeReference<Map<String, Object>>() {}
+                        ))
+                        .ifPresent(bodyMap::putAll);
                 }
             } else {
                 HttpPostRequestDecoder decoder =
@@ -206,7 +211,10 @@ public class HandlerService {
             throw new RuntimeException("UnSupported Method " + fullHttpRequest.method());
         }
 
-        byte[] requestBody = JsonUtils.serialize(bodyMap).getBytes(StandardCharsets.UTF_8);
+        byte[] requestBody = Optional.ofNullable(JsonUtils.serialize(bodyMap))
+            .map(s -> s.getBytes(StandardCharsets.UTF_8))
+            .orElse(new byte[0]);
+
         httpEventWrapper.setBody(requestBody);
 
         metrics.getSummaryMetrics().recordDecodeTimeCost(System.currentTimeMillis() - bodyDecodeStart);
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalSubscribeEventProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalSubscribeEventProcessor.java
index fe09d92f2..94b14d919 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalSubscribeEventProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalSubscribeEventProcessor.java
@@ -41,12 +41,13 @@ import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
 
 import java.util.ArrayList;
-import java.util.Arrays;
+import java.util.Collections;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 
 import org.slf4j.Logger;
@@ -119,8 +120,10 @@ public class LocalSubscribeEventProcessor extends AbstractEventProcessor impleme
         //validate body
         byte[] requestBody = requestWrapper.getBody();
 
-        Map<String, Object> requestBodyMap = JsonUtils.deserialize(new String(requestBody, Constants.DEFAULT_CHARSET),
-            new TypeReference<HashMap<String, Object>>() {});
+        Map<String, Object> requestBodyMap = Optional.ofNullable(JsonUtils.deserialize(
+            new String(requestBody, Constants.DEFAULT_CHARSET),
+            new TypeReference<HashMap<String, Object>>() {}
+        )).orElse(new HashMap<>());
 
         if (requestBodyMap.get("url") == null || requestBodyMap.get("topic") == null || requestBodyMap.get("consumerGroup") == null) {
             handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR, responseHeaderMap,
@@ -133,8 +136,10 @@ public class LocalSubscribeEventProcessor extends AbstractEventProcessor impleme
         String topic = JsonUtils.serialize(requestBodyMap.get("topic"));
 
         // SubscriptionItem
-        List<SubscriptionItem> subscriptionList = JsonUtils.deserialize(topic, new TypeReference<List<SubscriptionItem>>() {
-        });
+        List<SubscriptionItem> subscriptionList = Optional.ofNullable(JsonUtils.deserialize(
+            topic,
+            new TypeReference<List<SubscriptionItem>>() {}
+        )).orElse(Collections.emptyList());
 
         //do acl check
         if (eventMeshHTTPServer.getEventMeshHttpConfiguration().isEventMeshServerSecurityEnable()) {
@@ -213,7 +218,7 @@ public class LocalSubscribeEventProcessor extends AbstractEventProcessor impleme
                     consumeTopicConfig.setConsumerGroup(consumerGroup);
                     consumeTopicConfig.setTopic(subTopic.getTopic());
                     consumeTopicConfig.setSubscriptionItem(subTopic);
-                    consumeTopicConfig.setUrls(new HashSet<>(Arrays.asList(url)));
+                    consumeTopicConfig.setUrls(new HashSet<>(Collections.singletonList(url)));
 
                     consumeTopicConfig.setIdcUrls(idcUrls);
 
@@ -230,7 +235,7 @@ public class LocalSubscribeEventProcessor extends AbstractEventProcessor impleme
                         newTopicConf.setConsumerGroup(consumerGroup);
                         newTopicConf.setTopic(subTopic.getTopic());
                         newTopicConf.setSubscriptionItem(subTopic);
-                        newTopicConf.setUrls(new HashSet<>(Arrays.asList(url)));
+                        newTopicConf.setUrls(new HashSet<>(Collections.singletonList(url)));
                         newTopicConf.setIdcUrls(idcUrls);
                         map.put(subTopic.getTopic(), newTopicConf);
                     }
@@ -244,7 +249,7 @@ public class LocalSubscribeEventProcessor extends AbstractEventProcessor impleme
                         latestTopicConf.setConsumerGroup(consumerGroup);
                         latestTopicConf.setTopic(subTopic.getTopic());
                         latestTopicConf.setSubscriptionItem(subTopic);
-                        latestTopicConf.setUrls(new HashSet<>(Arrays.asList(url)));
+                        latestTopicConf.setUrls(new HashSet<>(Collections.singletonList(url)));
 
                         ConsumerGroupTopicConf currentTopicConf = set.getValue();
                         latestTopicConf.getUrls().addAll(currentTopicConf.getUrls());
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalUnSubscribeEventProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalUnSubscribeEventProcessor.java
index dff668f64..d19aa49d2 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalUnSubscribeEventProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalUnSubscribeEventProcessor.java
@@ -37,12 +37,14 @@ import org.apache.eventmesh.runtime.util.RemotingHelper;
 import org.apache.commons.lang3.StringUtils;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 
 import org.slf4j.Logger;
@@ -52,6 +54,7 @@ import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.http.HttpRequest;
 
 import com.fasterxml.jackson.core.type.TypeReference;
+import com.google.common.collect.Maps;
 
 @EventMeshTrace(isEnable = false)
 public class LocalUnSubscribeEventProcessor extends AbstractEventProcessor implements AsyncHttpProcessor {
@@ -112,9 +115,10 @@ public class LocalUnSubscribeEventProcessor extends AbstractEventProcessor imple
         //validate body
         byte[] requestBody = requestWrapper.getBody();
 
-        Map<String, Object> requestBodyMap = JsonUtils.deserialize(new String(requestBody, Constants.DEFAULT_CHARSET),
-            new TypeReference<HashMap<String, Object>>() {
-            });
+        Map<String, Object> requestBodyMap = Optional.ofNullable(JsonUtils.deserialize(
+            new String(requestBody, Constants.DEFAULT_CHARSET),
+            new TypeReference<HashMap<String, Object>>() {}
+        )).orElse(Maps.newHashMap());
 
         if (requestBodyMap.get(EventMeshConstants.URL) == null
             || requestBodyMap.get(EventMeshConstants.MANAGE_TOPIC) == null
@@ -129,8 +133,10 @@ public class LocalUnSubscribeEventProcessor extends AbstractEventProcessor imple
         String topic = JsonUtils.serialize(requestBodyMap.get(EventMeshConstants.MANAGE_TOPIC));
 
         // unSubscriptionItem
-        List<String> unSubTopicList = JsonUtils.deserialize(topic, new TypeReference<List<String>>() {
-        });
+        List<String> unSubTopicList = Optional.ofNullable(JsonUtils.deserialize(
+            topic,
+            new TypeReference<List<String>>() {}
+        )).orElse(Collections.emptyList());
 
         String env = sysHeaderMap.get(ProtocolKey.ClientInstanceKey.ENV).toString();
         String idc = sysHeaderMap.get(ProtocolKey.ClientInstanceKey.IDC).toString();
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/RemoteSubscribeEventProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/RemoteSubscribeEventProcessor.java
index 83c5c97cd..96cc3c37a 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/RemoteSubscribeEventProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/RemoteSubscribeEventProcessor.java
@@ -46,9 +46,11 @@ import org.apache.http.impl.client.CloseableHttpClient;
 import org.apache.http.util.EntityUtils;
 
 import java.io.IOException;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -58,6 +60,7 @@ import io.netty.handler.codec.http.HttpRequest;
 
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.google.common.base.Preconditions;
+import com.google.common.collect.Maps;
 
 @EventMeshTrace(isEnable = false)
 public class RemoteSubscribeEventProcessor extends AbstractEventProcessor implements AsyncHttpProcessor {
@@ -122,9 +125,10 @@ public class RemoteSubscribeEventProcessor extends AbstractEventProcessor implem
         //validate body
         byte[] requestBody = requestWrapper.getBody();
 
-        Map<String, Object> requestBodyMap = JsonUtils.deserialize(new String(requestBody, Constants.DEFAULT_CHARSET),
-            new TypeReference<HashMap<String, Object>>() {
-            });
+        Map<String, Object> requestBodyMap = Optional.ofNullable(JsonUtils.deserialize(
+            new String(requestBody, Constants.DEFAULT_CHARSET),
+            new TypeReference<HashMap<String, Object>>() {}
+        )).orElse(Maps.newHashMap());
 
 
         if (requestBodyMap.get(EventMeshConstants.URL) == null
@@ -141,8 +145,10 @@ public class RemoteSubscribeEventProcessor extends AbstractEventProcessor implem
 
 
         // SubscriptionItem
-        List<SubscriptionItem> subscriptionList = JsonUtils.deserialize(topic, new TypeReference<List<SubscriptionItem>>() {
-        });
+        List<SubscriptionItem> subscriptionList = Optional.ofNullable(JsonUtils.deserialize(
+            topic,
+            new TypeReference<List<SubscriptionItem>>() {}
+        )).orElse(Collections.emptyList());
 
         //do acl check
         if (eventMeshHTTPServer.getEventMeshHttpConfiguration().isEventMeshServerSecurityEnable()) {
@@ -236,8 +242,10 @@ public class RemoteSubscribeEventProcessor extends AbstractEventProcessor implem
             String remoteResult = post(closeableHttpClient, targetMesh, remoteHeaderMap, remoteBodyMap,
                 response -> EntityUtils.toString(response.getEntity(), Constants.DEFAULT_CHARSET));
 
-            Map<String, String> remoteResultMap = JsonUtils.deserialize(remoteResult, new TypeReference<Map<String, String>>() {
-            });
+            Map<String, String> remoteResultMap = Optional.ofNullable(JsonUtils.deserialize(
+                remoteResult,
+                new TypeReference<Map<String, String>>() {}
+            )).orElse(Maps.newHashMap());
 
             if (String.valueOf(EventMeshRetCode.SUCCESS.getRetCode()).equals(remoteResultMap.get(EventMeshConstants.RET_CODE))) {
                 responseBodyMap.put(EventMeshConstants.RET_CODE, EventMeshRetCode.SUCCESS.getRetCode());
@@ -286,7 +294,7 @@ public class RemoteSubscribeEventProcessor extends AbstractEventProcessor implem
 
         //body
         if (MapUtils.isNotEmpty(requestBody)) {
-            String jsonStr = JsonUtils.serialize(requestBody);
+            String jsonStr = Optional.ofNullable(JsonUtils.serialize(requestBody)).orElse("");
             httpPost.setEntity(new StringEntity(jsonStr, ContentType.APPLICATION_JSON));
         }
 
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/RemoteUnSubscribeEventProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/RemoteUnSubscribeEventProcessor.java
index 8c29ddcc0..6e61c5b47 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/RemoteUnSubscribeEventProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/RemoteUnSubscribeEventProcessor.java
@@ -44,10 +44,12 @@ import org.apache.http.impl.client.CloseableHttpClient;
 import org.apache.http.util.EntityUtils;
 
 import java.io.IOException;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.stream.Collectors;
 
 import org.slf4j.Logger;
@@ -58,6 +60,7 @@ import io.netty.handler.codec.http.HttpRequest;
 
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.google.common.base.Preconditions;
+import com.google.common.collect.Maps;
 
 @EventMeshTrace(isEnable = false)
 public class RemoteUnSubscribeEventProcessor extends AbstractEventProcessor implements AsyncHttpProcessor {
@@ -119,9 +122,10 @@ public class RemoteUnSubscribeEventProcessor extends AbstractEventProcessor impl
         //validate body
         byte[] requestBody = requestWrapper.getBody();
 
-        Map<String, Object> requestBodyMap = JsonUtils.deserialize(new String(requestBody, Constants.DEFAULT_CHARSET),
-            new TypeReference<HashMap<String, Object>>() {
-            });
+        Map<String, Object> requestBodyMap = Optional.ofNullable(JsonUtils.deserialize(
+            new String(requestBody, Constants.DEFAULT_CHARSET),
+            new TypeReference<HashMap<String, Object>>() {}
+        )).orElse(Maps.newHashMap());
 
         if (requestBodyMap.get(EventMeshConstants.URL) == null
             || requestBodyMap.get(EventMeshConstants.MANAGE_TOPIC) == null
@@ -165,9 +169,10 @@ public class RemoteUnSubscribeEventProcessor extends AbstractEventProcessor impl
             remoteBodyMap.put(EventMeshConstants.CONSUMER_GROUP, meshGroup);
             remoteBodyMap.put(EventMeshConstants.MANAGE_TOPIC, requestBodyMap.get(EventMeshConstants.MANAGE_TOPIC));
 
-            List<String> unSubTopicList =
-                JsonUtils.deserialize(JsonUtils.serialize(requestBodyMap.get(EventMeshConstants.MANAGE_TOPIC)), new TypeReference<List<String>>() {
-                });
+            List<String> unSubTopicList = Optional.ofNullable(JsonUtils.deserialize(
+                JsonUtils.serialize(requestBodyMap.get(EventMeshConstants.MANAGE_TOPIC)),
+                new TypeReference<List<String>>() {}
+            )).orElse(Collections.emptyList());
 
             String targetMesh = "";
             if (!Objects.isNull(requestBodyMap.get("remoteMesh"))) {
@@ -190,8 +195,10 @@ public class RemoteUnSubscribeEventProcessor extends AbstractEventProcessor impl
             String remoteResult = post(closeableHttpClient, targetMesh, remoteHeaderMap, remoteBodyMap,
                 response -> EntityUtils.toString(response.getEntity(), Constants.DEFAULT_CHARSET));
 
-            Map<String, String> remoteResultMap = JsonUtils.deserialize(remoteResult, new TypeReference<Map<String, String>>() {
-            });
+            Map<String, String> remoteResultMap = Optional.ofNullable(JsonUtils.deserialize(
+                remoteResult,
+                new TypeReference<Map<String, String>>() {}
+            )).orElse(Maps.newHashMap());
 
             if (String.valueOf(EventMeshRetCode.SUCCESS.getRetCode()).equals(remoteResultMap.get(EventMeshConstants.RET_CODE))) {
                 responseBodyMap.put(EventMeshConstants.RET_CODE, EventMeshRetCode.SUCCESS.getRetCode());
@@ -238,7 +245,7 @@ public class RemoteUnSubscribeEventProcessor extends AbstractEventProcessor impl
 
         //body
         if (MapUtils.isNotEmpty(requestBody)) {
-            String jsonStr = JsonUtils.serialize(requestBody);
+            String jsonStr = Optional.ofNullable(JsonUtils.serialize(requestBody)).orElse("");
             httpPost.setEntity(new StringEntity(jsonStr, ContentType.APPLICATION_JSON));
         }
 
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ReplyMessageProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ReplyMessageProcessor.java
index b201f7158..82492b5f3 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ReplyMessageProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ReplyMessageProcessor.java
@@ -48,6 +48,7 @@ import org.apache.commons.lang3.StringUtils;
 
 import java.nio.charset.StandardCharsets;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 
 import org.slf4j.Logger;
@@ -182,8 +183,9 @@ public class ReplyMessageProcessor implements HttpRequestProcessor {
 
         String origTopic = event.getSubject();
 
-        final String replyMQCluster = event.getExtension(EventMeshConstants.PROPERTY_MESSAGE_CLUSTER) == null ? "" :
-            event.getExtension(EventMeshConstants.PROPERTY_MESSAGE_CLUSTER).toString();
+        final String replyMQCluster = Optional.ofNullable(event.getExtension(EventMeshConstants.PROPERTY_MESSAGE_CLUSTER))
+            .map(Objects::toString)
+            .orElse("");
         if (!org.apache.commons.lang3.StringUtils.isEmpty(replyMQCluster)) {
             replyTopic = replyMQCluster + "-" + replyTopic;
         } else {
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncRemoteEventProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncRemoteEventProcessor.java
index 81f062374..ed1726e71 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncRemoteEventProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncRemoteEventProcessor.java
@@ -48,6 +48,8 @@ import java.nio.charset.StandardCharsets;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 
 import org.slf4j.Logger;
@@ -60,6 +62,7 @@ import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.http.HttpRequest;
 
 import com.fasterxml.jackson.core.type.TypeReference;
+import com.google.common.collect.Maps;
 
 @EventMeshTrace(isEnable = true)
 public class SendAsyncRemoteEventProcessor implements AsyncHttpProcessor {
@@ -70,7 +73,7 @@ public class SendAsyncRemoteEventProcessor implements AsyncHttpProcessor {
 
     public Logger aclLogger = LoggerFactory.getLogger(EventMeshConstants.ACL);
 
-    private EventMeshHTTPServer eventMeshHTTPServer;
+    private final EventMeshHTTPServer eventMeshHTTPServer;
 
     public SendAsyncRemoteEventProcessor(EventMeshHTTPServer eventMeshHTTPServer) {
         this.eventMeshHTTPServer = eventMeshHTTPServer;
@@ -114,9 +117,10 @@ public class SendAsyncRemoteEventProcessor implements AsyncHttpProcessor {
         requestWrapper.buildSysHeaderForCE();
 
         // process remote event body
-        Map<String, Object> bodyMap = JsonUtils.deserialize(new String(requestWrapper.getBody(), Constants.DEFAULT_CHARSET),
-                new TypeReference<Map<String, Object>>() {
-                });
+        Map<String, Object> bodyMap = Optional.ofNullable(JsonUtils.deserialize(
+            new String(requestWrapper.getBody(), Constants.DEFAULT_CHARSET),
+            new TypeReference<Map<String, Object>>() {}
+        )).orElse(Maps.newHashMap());
 
         byte[] convertedBody = bodyMap.get("content").toString().getBytes(StandardCharsets.UTF_8);
         requestWrapper.setBody(convertedBody);
@@ -174,12 +178,9 @@ public class SendAsyncRemoteEventProcessor implements AsyncHttpProcessor {
             return;
         }
 
-        idc = event.getExtension(ProtocolKey.ClientInstanceKey.IDC) == null ? "" :
-                event.getExtension(ProtocolKey.ClientInstanceKey.IDC).toString();
-        String pid = event.getExtension(ProtocolKey.ClientInstanceKey.PID) == null ? "" :
-                event.getExtension(ProtocolKey.ClientInstanceKey.PID).toString();
-        String sys = event.getExtension(ProtocolKey.ClientInstanceKey.SYS) == null ? "" :
-                event.getExtension(ProtocolKey.ClientInstanceKey.SYS).toString();
+        idc = getExtension(event, ProtocolKey.ClientInstanceKey.IDC);
+        String pid = getExtension(event, ProtocolKey.ClientInstanceKey.PID);
+        String sys = getExtension(event, ProtocolKey.ClientInstanceKey.SYS);
 
         //validate event-extension
         if (StringUtils.isBlank(idc)
@@ -191,8 +192,7 @@ public class SendAsyncRemoteEventProcessor implements AsyncHttpProcessor {
             return;
         }
 
-        String producerGroup = event.getExtension(ProtocolKey.ClientInstanceKey.PRODUCERGROUP) == null ? "" :
-                event.getExtension(ProtocolKey.ClientInstanceKey.PRODUCERGROUP).toString();
+        String producerGroup = getExtension(event, ProtocolKey.ClientInstanceKey.PRODUCERGROUP);
         String topic = event.getSubject();
 
         //validate body
@@ -209,13 +209,9 @@ public class SendAsyncRemoteEventProcessor implements AsyncHttpProcessor {
         //do acl check
         if (eventMeshHTTPServer.getEventMeshHttpConfiguration().isEventMeshServerSecurityEnable()) {
             String remoteAddr = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
-            String user = event.getExtension(ProtocolKey.ClientInstanceKey.USERNAME) == null ? "" :
-                    event.getExtension(ProtocolKey.ClientInstanceKey.USERNAME).toString();
-            String pass = event.getExtension(ProtocolKey.ClientInstanceKey.PASSWD) == null ? "" :
-                    event.getExtension(ProtocolKey.ClientInstanceKey.PASSWD).toString();
-            String subsystem =
-                    event.getExtension(ProtocolKey.ClientInstanceKey.SYS) == null ? "" :
-                            event.getExtension(ProtocolKey.ClientInstanceKey.SYS).toString();
+            String user = getExtension(event, ProtocolKey.ClientInstanceKey.USERNAME);
+            String pass = getExtension(event, ProtocolKey.ClientInstanceKey.PASSWD);
+            String subsystem = getExtension(event, ProtocolKey.ClientInstanceKey.SYS);
             String requestURI = requestWrapper.getRequestURI();
             try {
                 Acl.doAclCheckInHttpSend(remoteAddr, user, pass, subsystem, topic, requestURI);
@@ -319,6 +315,12 @@ public class SendAsyncRemoteEventProcessor implements AsyncHttpProcessor {
         }
     }
 
+    private String getExtension(CloudEvent event, String protocolKey) {
+        return Optional.ofNullable(event.getExtension(protocolKey))
+            .map(Objects::toString)
+            .orElse("");
+    }
+
     @Override
     public String[] paths() {
         return new String[]{RequestURI.PUBLISH_BRIDGE.getRequestURI()};
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/inf/AbstractEventProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/inf/AbstractEventProcessor.java
index da0aa72ea..08a66752d 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/inf/AbstractEventProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/inf/AbstractEventProcessor.java
@@ -35,10 +35,13 @@ import org.apache.commons.lang3.StringUtils;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.google.common.collect.Maps;
+
 /**
  * EventProcessor
  */
@@ -121,7 +124,9 @@ public class AbstractEventProcessor {
             ConsumerGroupMetadata consumerGroupMetadata =
                     JsonUtils.deserialize(topicMetadataJson, ConsumerGroupMetadata.class);
             Map<String, ConsumerGroupTopicMetadata> consumerGroupTopicMetadataMap =
-                    consumerGroupMetadata.getConsumerGroupTopicMetadataMap();
+                    Optional.ofNullable(consumerGroupMetadata)
+                        .map(ConsumerGroupMetadata::getConsumerGroupTopicMetadataMap)
+                        .orElse(Maps.newConcurrentMap());
 
             for (SubscriptionItem subscriptionItem : subscriptionList) {
                 if (consumerGroupTopicMetadataMap.containsKey(subscriptionItem.getTopic())) {