fix(proxy): fix wrong nums of StartAndShutdown

apache · Oct 23, 2024 · 230eb02 · 230eb02
1 parent 221d9a0
commit 230eb02
Show file tree

Hide file tree

Showing 3 changed files with 14 additions and 4 deletions.
diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServerBuilder.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServerBuilder.java
@@ -48,7 +48,7 @@ public GrpcServer build() {
         int port = ConfigurationManager.getProxyConfig().getGrpcServerPort();
         //
         ThreadPoolExecutor executor = createServerExecutor();
-        initializer.getStartAndShutdowns().add(new StartAndShutdown() {
+        appendStartAndShutdown(new StartAndShutdown() {
             @Override
             public void shutdown() throws Exception {
                 executor.shutdown();
@@ -61,7 +61,7 @@ public void start() throws Exception {
         });
         //
         GrpcMessagingApplication application = GrpcMessagingApplication.create(initializer.getMessagingProcessor());
-        initializer.getStartAndShutdowns().add(application);
+        appendStartAndShutdown(application);
         //
         NettyServerBuilder serverBuilder = NettyServerBuilder.forPort(port);
         serverBuilder.addService(application)

diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/spi/ProxyServerFactoryBase.java b/proxy/src/main/java/org/apache/rocketmq/proxy/spi/ProxyServerFactoryBase.java
@@ -18,13 +18,16 @@
 package org.apache.rocketmq.proxy.spi;
 
 import org.apache.rocketmq.acl.AccessValidator;
+import org.apache.rocketmq.common.utils.StartAndShutdown;
 
+import java.util.ArrayList;
 import java.util.List;
 
 public abstract class ProxyServerFactoryBase implements ProxyServerFactory {
 
     protected List<AccessValidator> validators;
     protected ProxyServerInitializer initializer;
+    protected final List<StartAndShutdown> startAndShutdowns = new ArrayList<StartAndShutdown>();
 
     @Override
     public ProxyServerFactory withAccessValidators(List<AccessValidator> accessValidators) {
@@ -41,11 +44,18 @@ public ProxyServerFactory withInitializer(ProxyServerInitializer proxyServerInit
     @Override
     public final ProxyServerBase get() {
         ProxyServerBase serverBase = build();
+        this.initializer.getStartAndShutdowns().forEach(this::appendStartAndShutdown);
         serverBase.setBrokerController(this.initializer.getBrokerController());
-        serverBase.setStartAndShutdowns(this.initializer.getStartAndShutdowns());
+        serverBase.setStartAndShutdowns(this.startAndShutdowns);
         serverBase.setMessagingProcessor(this.initializer.getMessagingProcessor());
         return serverBase;
     }
 
+    protected void appendStartAndShutdown(StartAndShutdown sas) {
+        if (sas != null && !this.startAndShutdowns.contains(sas)) {
+            this.startAndShutdowns.add(sas);
+        }
+    }
+
     protected abstract ProxyServerBase build();
 }
diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/spi/ProxyServerInitializer.java b/proxy/src/main/java/org/apache/rocketmq/proxy/spi/ProxyServerInitializer.java
@@ -110,7 +110,7 @@ private void initializeConfiguration() throws Exception {
 
     private void validateConfiguration() throws Exception {
         String proxyModeStr = ConfigurationManager.getProxyConfig().getProxyMode();
-        if (!ProxyMode.isClusterMode(proxyModeStr) || !ProxyMode.isLocalMode(proxyModeStr)) {
+        if (!ProxyMode.isClusterMode(proxyModeStr) && !ProxyMode.isLocalMode(proxyModeStr)) {
             throw new IllegalArgumentException("try to start proxy server with wrong mode, use 'local' or 'cluster'");
         }
     }