Repository: apache/eventmesh
Issue: [Bug]ProducerTopicManager throw NPE when runtime module started

### 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

Windows

### EventMesh version

master

### What happened

when I start eventmesh runtime module. when ProducerTopicManager.scheduler execute then will throw java.lang.NullPointerException:

```
2023-03-12 12:27:53,048 INFO  [eventMesh-metrics-2] HTTPMetricsServer(HTTPMetricsServer.java:155) - maxPushMsgTPS: 0.0, avgPushMsgTPS: 0.0, sum: 0, sumFail: 0, sumFailRate: 0.0, maxClientLatency: 0.0, avgClientLatency: 0.0
2023-03-12 12:27:53,048 INFO  [eventMesh-metrics-2] HTTPMetricsServer(HTTPMetricsServer.java:170) - batchMsgQ: 0, sendMsgQ: 0, pushMsgQ: 0, httpRetryQ: 0
2023-03-12 12:27:53,048 INFO  [eventMesh-metrics-2] HTTPMetricsServer(HTTPMetricsServer.java:178) - batchAvgSend2MQCost: 0.0, avgSend2MQCost: 0.0, avgReply2MQCost: 0.0
java.lang.NullPointerException
	at org.apache.eventmesh.runtime.core.protocol.http.producer.ProducerTopicManager.lambda$start$1(ProducerTopicManager.java:69)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.runAndReset$$$capture(FutureTask.java:308)
	at java.util.concurrent.FutureTask.runAndReset(FutureTask.java)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
2023-03-12 12:28:02,065 DEBUG [pool-4-thread-1] ConsumerManager(ConsumerManager.java:214) - grpc client info check
2023-03-12 12:28:02,065 DEBUG [pool-4-thread-1] ConsumerManager(ConsumerManager.java:223) - total number of ConsumerGroupClients: 0
2023-03-12 12:28:12,060 DEBUG [pool-4-thread-1] ConsumerManager(ConsumerManager.java:214) - grpc client info check
2023-03-12 12:28:12,060 DEBUG [pool-4-thread-1] ConsumerManager(ConsumerManager.java:223) - total number of ConsumerGroupClients: 0
java.lang.NullPointerException
	at org.apache.eventmesh.runtime.core.protocol.http.producer.ProducerTopicManager.lambda$start$1(ProducerTopicManager.java:69)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.runAndReset$$$capture(FutureTask.java:308)
	at java.util.concurrent.FutureTask.runAndReset(FutureTask.java)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
2023-03-12 12:28:22,063 DEBUG [pool-4-thread-1] ConsumerManager(ConsumerManager.java:214) - grpc client info check
2023-03-12 12:28:22,063 DEBUG [pool-4-thread-1] ConsumerManager(ConsumerManager.java:223) - total number of ConsumerGroupClients: 0
```


### How to reproduce

start eventmesh runtime

### Debug logs

```
2023-03-12 12:27:53,048 INFO  [eventMesh-metrics-2] HTTPMetricsServer(HTTPMetricsServer.java:155) - maxPushMsgTPS: 0.0, avgPushMsgTPS: 0.0, sum: 0, sumFail: 0, sumFailRate: 0.0, maxClientLatency: 0.0, avgClientLatency: 0.0
2023-03-12 12:27:53,048 INFO  [eventMesh-metrics-2] HTTPMetricsServer(HTTPMetricsServer.java:170) - batchMsgQ: 0, sendMsgQ: 0, pushMsgQ: 0, httpRetryQ: 0
2023-03-12 12:27:53,048 INFO  [eventMesh-metrics-2] HTTPMetricsServer(HTTPMetricsServer.java:178) - batchAvgSend2MQCost: 0.0, avgSend2MQCost: 0.0, avgReply2MQCost: 0.0
java.lang.NullPointerException
	at org.apache.eventmesh.runtime.core.protocol.http.producer.ProducerTopicManager.lambda$start$1(ProducerTopicManager.java:69)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.runAndReset$$$capture(FutureTask.java:308)
	at java.util.concurrent.FutureTask.runAndReset(FutureTask.java)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
2023-03-12 12:28:02,065 DEBUG [pool-4-thread-1] ConsumerManager(ConsumerManager.java:214) - grpc client info check
2023-03-12 12:28:02,065 DEBUG [pool-4-thread-1] ConsumerManager(ConsumerManager.java:223) - total number of ConsumerGroupClients: 0
2023-03-12 12:28:12,060 DEBUG [pool-4-thread-1] ConsumerManager(ConsumerManager.java:214) - grpc client info check
2023-03-12 12:28:12,060 DEBUG [pool-4-thread-1] ConsumerManager(ConsumerManager.java:223) - total number of ConsumerGroupClients: 0
java.lang.NullPointerException
	at org.apache.eventmesh.runtime.core.protocol.http.producer.ProducerTopicManager.lambda$start$1(ProducerTopicManager.java:69)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.runAndReset$$$capture(FutureTask.java:308)
	at java.util.concurrent.FutureTask.runAndReset(FutureTask.java)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
2023-03-12 12:28:22,063 DEBUG [pool-4-thread-1] ConsumerManager(ConsumerManager.java:214) - grpc client info check
2023-03-12 12:28:22,063 DEBUG [pool-4-thread-1] ConsumerManager(ConsumerManager.java:223) - total number of ConsumerGroupClients: 0
```

### 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/producer/ProducerTopicManager.java
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshServer.java
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/Session.java

diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshServer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshServer.java
index dfbad0a38..5d0fd427a 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshServer.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshServer.java
@@ -218,4 +218,8 @@ public class EventMeshServer {
     public ProducerTopicManager getProducerTopicManager() {
         return producerTopicManager;
     }
+
+    public CommonConfiguration getConfiguration() {
+        return configuration;
+    }
 }
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/producer/ProducerTopicManager.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/producer/ProducerTopicManager.java
index 785540424..5568cad59 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/producer/ProducerTopicManager.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/producer/ProducerTopicManager.java
@@ -23,39 +23,33 @@ import org.apache.eventmesh.common.ThreadPoolFactory;
 import org.apache.eventmesh.runtime.boot.EventMeshServer;
 
 import java.util.List;
+import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
+import lombok.extern.slf4j.Slf4j;
 
+@Slf4j
 public class ProducerTopicManager {
 
-    private Logger retryLogger = LoggerFactory.getLogger("p-topic-m");
-
-    private Logger logger = LoggerFactory.getLogger(this.getClass());
-
     private EventMeshServer eventMeshServer;
 
-    public ProducerTopicManager(EventMeshServer eventMeshServer) {
-        this.eventMeshServer = eventMeshServer;
-    }
-
     private transient ScheduledFuture<?> scheduledTask;
 
     protected static ScheduledExecutorService scheduler;
 
-    private ConcurrentHashMap<String, EventMeshServicePubTopicInfo> eventMeshServicePubTopicInfoMap = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<String, EventMeshServicePubTopicInfo> eventMeshServicePubTopicInfoMap = new ConcurrentHashMap<>(64);
 
-    public void init() {
+    public ProducerTopicManager(EventMeshServer eventMeshServer) {
+        this.eventMeshServer = eventMeshServer;
+    }
 
+    public void init() {
         scheduler = ThreadPoolFactory.createScheduledExecutor(Runtime.getRuntime().availableProcessors(),
             new EventMeshThreadFactory("Producer-Topic-Manager", true));
-        logger.info("ProducerTopicManager inited......");
-
+        log.info("ProducerTopicManager inited......");
     }
 
     public void start() {
@@ -63,29 +57,28 @@ public class ProducerTopicManager {
         if (scheduledTask == null) {
             synchronized (ProducerTopicManager.class) {
                 scheduledTask = scheduler.scheduleAtFixedRate(() -> {
-
                     try {
-                        List<EventMeshServicePubTopicInfo> list = eventMeshServer.getRegistry().findEventMeshServicePubTopicInfos();
-                        list.forEach(e -> {
-                            eventMeshServicePubTopicInfoMap.put(e.getService(), e);
-                        });
+                        if (!eventMeshServer.getConfiguration().isEventMeshServerRegistryEnable()) {
+                            return;
+                        }
+                        List<EventMeshServicePubTopicInfo> pubTopicInfoList = eventMeshServer.getRegistry().findEventMeshServicePubTopicInfos();
+                        Optional.ofNullable(pubTopicInfoList)
+                            .ifPresent(lt -> lt.forEach(item -> eventMeshServicePubTopicInfoMap.put(item.getService(), item)));
                     } catch (Exception e) {
-                        e.printStackTrace();
+                        log.error("ProducerTopicManager update eventMesh pub topic info error. ", e);
                     }
 
                 }, 5, 20, TimeUnit.SECONDS);
             }
         }
-
-
-        logger.info("ProducerTopicManager started......");
+        log.info("ProducerTopicManager started......");
     }
 
     public void shutdown() {
         if (scheduledTask != null) {
             scheduledTask.cancel(false);
         }
-        logger.info("ProducerTopicManager shutdown......");
+        log.info("ProducerTopicManager shutdown......");
     }
 
     public ConcurrentHashMap<String, EventMeshServicePubTopicInfo> getEventMeshServicePubTopicInfoMap() {
@@ -97,4 +90,4 @@ public class ProducerTopicManager {
     }
 
 
-}
\\ No newline at end of file
+}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/Session.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/Session.java
index b76339c1d..9030fae02 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/Session.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/Session.java
@@ -326,19 +326,19 @@ public class Session {
 
     public void trySendListenResponse(Header header, long startTime, long taskExecuteTime) {
         if (!listenRspSend && listenRspLock.tryLock()) {
-                if (!listenRspSend) {
-                    if (header == null) {
-                        header = new Header(LISTEN_RESPONSE, OPStatus.SUCCESS.getCode(), "succeed", null);
-                    }
-                    Package msg = new Package();
-                    msg.setHeader(header);
-
-                    // TODO: if startTime is modified
-                    Utils.writeAndFlush(msg, startTime, taskExecuteTime, context, this);
-                    listenRspSend = true;
+            if (!listenRspSend) {
+                if (header == null) {
+                    header = new Header(LISTEN_RESPONSE, OPStatus.SUCCESS.getCode(), "succeed", null);
                 }
-                listenRspLock.unlock();
-            
+                Package msg = new Package();
+                msg.setHeader(header);
+
+                // TODO: if startTime is modified
+                Utils.writeAndFlush(msg, startTime, taskExecuteTime, context, this);
+                listenRspSend = true;
+            }
+            listenRspLock.unlock();
+
         }
     }