[ISSUE #3949] v2 support

proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java

@@ -21,6 +21,8 @@ import ch.qos.logback.classic.LoggerContext;
 import ch.qos.logback.classic.joran.JoranConfigurator;
 import ch.qos.logback.core.joran.spi.JoranException;
 import java.util.Date;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 import org.apache.rocketmq.broker.BrokerController;
 import org.apache.rocketmq.broker.BrokerStartup;
 import org.apache.rocketmq.client.log.ClientLogger;
@@ -33,6 +35,8 @@ import org.apache.rocketmq.proxy.common.StartAndShutdown;
 import org.apache.rocketmq.proxy.config.ConfigurationManager;
 import org.apache.rocketmq.proxy.config.ProxyConfig;
 import org.apache.rocketmq.proxy.grpc.GrpcServer;
+import org.apache.rocketmq.proxy.grpc.GrpcServerBuilder;
+import org.apache.rocketmq.proxy.grpc.v2.GrpcMessagingProcessor;
 import org.apache.rocketmq.proxy.grpc.v2.adapter.ProxyMode;
 import org.apache.rocketmq.proxy.grpc.v2.service.ClusterGrpcService;
 import org.apache.rocketmq.proxy.grpc.v2.service.GrpcForwardService;
@@ -59,8 +63,12 @@ public class ProxyStartup {
             // init thread pool monitor for proxy.
             initThreadPoolMonitor();
 
+            ThreadPoolExecutor executor = createServerExecutor();
+
             // create grpcServer
-            GrpcServer grpcServer = createGrpcServer();
+            GrpcServer grpcServer = GrpcServerBuilder.newBuilder(executor)
+                .addService(createServiceProcessor())
+                .build();
             PROXY_START_AND_SHUTDOWN.appendStartAndShutdown(grpcServer);
 
             // start servers one by one.
@@ -85,7 +93,7 @@ public class ProxyStartup {
         log.info(new Date() + " rmq-proxy startup successfully");
     }
 
-    private static GrpcServer createGrpcServer() throws Exception {
+    private static GrpcMessagingProcessor createServiceProcessor() {
         GrpcForwardService grpcService;
         String proxyModeStr = ConfigurationManager.getProxyConfig().getProxyMode();
         if (ProxyMode.isClusterMode(proxyModeStr)) {
@@ -109,7 +117,8 @@ public class ProxyStartup {
             throw new IllegalArgumentException("try to start grpc server with wrong mode, use 'local' or 'cluster'");
         }
 
-        return new GrpcServer(grpcService);
+        PROXY_START_AND_SHUTDOWN.appendStartAndShutdown(grpcService);
+        return new GrpcMessagingProcessor(grpcService);
     }
 
     private static BrokerController createBrokerController() {
@@ -117,6 +126,20 @@ public class ProxyStartup {
         return BrokerStartup.createBrokerController(brokerStartupArgs);
     }
 
+    public static ThreadPoolExecutor createServerExecutor() {
+        int threadPoolNums = ConfigurationManager.getProxyConfig().getGrpcThreadPoolNums();
+        int threadPoolQueueCapacity = ConfigurationManager.getProxyConfig().getGrpcThreadPoolQueueCapacity();
+        ThreadPoolExecutor executor = ThreadPoolMonitor.createAndMonitor(
+            threadPoolNums,
+            threadPoolNums,
+            1, TimeUnit.MINUTES,
+            "GrpcRequestExecutorThread",
+            threadPoolQueueCapacity
+        );
+        PROXY_START_AND_SHUTDOWN.appendShutdown(executor::shutdown);
+        return executor;
+    }
+
     public static void initThreadPoolMonitor() {
         ThreadPoolMonitor.init();
         ProxyConfig config = ConfigurationManager.getProxyConfig();

proxy/src/main/java/org/apache/rocketmq/proxy/common/AbstractStartAndShutdown.java

@@ -41,4 +41,32 @@ public abstract class AbstractStartAndShutdown implements StartAndShutdown {
             startAndShutdownList.get(index).shutdown();
         }
     }
+
+    public void appendStart(Start start) {
+        this.appendStartAndShutdown(new StartAndShutdown() {
+            @Override
+            public void shutdown() throws Exception {
+
+            }
+
+            @Override
+            public void start() throws Exception {
+                start.start();
+            }
+        });
+    }
+
+    public void appendShutdown(Shutdown shutdown) {
+        this.appendStartAndShutdown(new StartAndShutdown() {
+            @Override
+            public void shutdown() throws Exception {
+                shutdown.shutdown();
+            }
+
+            @Override
+            public void start() throws Exception {
+
+            }
+        });
+    }
 }

proxy/src/main/java/org/apache/rocketmq/proxy/common/Shutdown.java

@@ -0,0 +1,22 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.proxy.common;
+
+public interface Shutdown {
+    void shutdown() throws Exception;
+}

proxy/src/main/java/org/apache/rocketmq/proxy/common/Start.java

@@ -0,0 +1,22 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.proxy.common;
+
+public interface Start {
+    void start() throws Exception;
+}

proxy/src/main/java/org/apache/rocketmq/proxy/common/StartAndShutdown.java

@@ -17,7 +17,5 @@
 
 package org.apache.rocketmq.proxy.common;
 
-public interface StartAndShutdown {
-    void start() throws Exception;
-    void shutdown() throws Exception;
+public interface StartAndShutdown extends Start, Shutdown {
 }

proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServer.java

@@ -17,152 +17,22 @@
 
 package org.apache.rocketmq.proxy.grpc;
 
-import io.grpc.netty.shaded.io.grpc.netty.GrpcSslContexts;
-import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder;
-import io.grpc.netty.shaded.io.netty.channel.epoll.EpollEventLoopGroup;
-import io.grpc.netty.shaded.io.netty.channel.epoll.EpollServerSocketChannel;
-import io.grpc.netty.shaded.io.netty.channel.nio.NioEventLoopGroup;
-import io.grpc.netty.shaded.io.netty.channel.socket.nio.NioServerSocketChannel;
-import io.grpc.netty.shaded.io.netty.handler.ssl.ClientAuth;
-import io.grpc.netty.shaded.io.netty.handler.ssl.util.InsecureTrustManagerFactory;
-import io.grpc.netty.shaded.io.netty.handler.ssl.util.SelfSignedCertificate;
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.security.cert.CertificateException;
-import java.util.List;
-import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
-import javax.net.ssl.SSLException;
-import org.apache.rocketmq.acl.AccessValidator;
 import org.apache.rocketmq.common.constant.LoggerName;
-import org.apache.rocketmq.common.thread.ThreadPoolMonitor;
-import org.apache.rocketmq.common.utils.ServiceProvider;
 import org.apache.rocketmq.logging.InternalLogger;
 import org.apache.rocketmq.logging.InternalLoggerFactory;
 import org.apache.rocketmq.proxy.common.StartAndShutdown;
-import org.apache.rocketmq.proxy.config.ConfigurationManager;
-import org.apache.rocketmq.proxy.config.ProxyConfig;
-import org.apache.rocketmq.proxy.grpc.interceptor.AuthenticationInterceptor;
-import org.apache.rocketmq.proxy.grpc.interceptor.ContextInterceptor;
-import org.apache.rocketmq.proxy.grpc.interceptor.HeaderInterceptor;
-import org.apache.rocketmq.proxy.grpc.v2.GrpcMessagingProcessor;
-import org.apache.rocketmq.proxy.grpc.v2.service.GrpcForwardService;
 
 public class GrpcServer implements StartAndShutdown {
     private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
 
     private final io.grpc.Server server;
-    private final ThreadPoolExecutor executor;
-    private final GrpcForwardService grpcForwardService;
 
-    public GrpcServer(GrpcForwardService grpcForwardService) {
-        this.grpcForwardService = grpcForwardService;
-        int port = ConfigurationManager.getProxyConfig().getGrpcServerPort();
-        NettyServerBuilder serverBuilder = NettyServerBuilder.forPort(port);
-
-        try {
-            configSslContext(serverBuilder);
-        } catch (Exception e) {
-            log.error("grpc tls set failed. msg: {}, e:", e.getMessage(), e);
-            throw new RuntimeException("grpc tls set failed: " + e.getMessage());
-        }
-
-        // create executor
-        int threadPoolNums = ConfigurationManager.getProxyConfig().getGrpcThreadPoolNums();
-        int threadPoolQueueCapacity = ConfigurationManager.getProxyConfig().getGrpcThreadPoolQueueCapacity();
-        this.executor = ThreadPoolMonitor.createAndMonitor(
-            threadPoolNums,
-            threadPoolNums,
-            1, TimeUnit.MINUTES,
-            "GrpcRequestExecutorThread",
-            threadPoolQueueCapacity
-        );
-
-        GrpcMessagingProcessor messagingProcessor = createProcessor();
-
-        // build server
-        int bossLoopNum = ConfigurationManager.getProxyConfig().getGrpcBossLoopNum();
-        int workerLoopNum = ConfigurationManager.getProxyConfig().getGrpcWorkerLoopNum();
-        int maxInboundMessageSize = ConfigurationManager.getProxyConfig().getGrpcMaxInboundMessageSize();
-
-        if (ConfigurationManager.getProxyConfig().isEnableGrpcEpoll()) {
-            serverBuilder.maxInboundMessageSize(maxInboundMessageSize)
-                .bossEventLoopGroup(new EpollEventLoopGroup(bossLoopNum))
-                .workerEventLoopGroup(new EpollEventLoopGroup(workerLoopNum))
-                .channelType(EpollServerSocketChannel.class)
-                .addService(messagingProcessor)
-                .executor(this.executor);
-        } else {
-            serverBuilder.maxInboundMessageSize(maxInboundMessageSize)
-                .bossEventLoopGroup(new NioEventLoopGroup(bossLoopNum))
-                .workerEventLoopGroup(new NioEventLoopGroup(workerLoopNum))
-                .channelType(NioServerSocketChannel.class)
-                .addService(messagingProcessor)
-                .executor(this.executor);
-        }
-
-        configInterceptor(serverBuilder);
-        this.server = serverBuilder.build();
-
-        log.info(
-            "grpc server has built. port: {}, tlsKeyPath: {}, tlsCertPath: {}, threadPool: {}, queueCapacity: {}, "
-                + "boosLoop: {}, workerLoop: {}, maxInboundMessageSize: {}",
-            port, threadPoolNums, threadPoolQueueCapacity,
-            bossLoopNum, workerLoopNum, maxInboundMessageSize);
-    }
-
-    protected GrpcMessagingProcessor createProcessor() {
-        return new GrpcMessagingProcessor(grpcForwardService);
-    }
-
-    protected void configSslContext(NettyServerBuilder serverBuilder) throws SSLException, CertificateException {
-        if (null == serverBuilder) {
-            return;
-        }
-        ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig();
-        boolean tlsTestModeEnable = proxyConfig.isGrpcTlsTestModeEnable();
-        if (tlsTestModeEnable) {
-            SelfSignedCertificate selfSignedCertificate = new SelfSignedCertificate();
-            serverBuilder.sslContext(GrpcSslContexts.forServer(selfSignedCertificate.certificate(), selfSignedCertificate.privateKey())
-                .trustManager(InsecureTrustManagerFactory.INSTANCE)
-                .clientAuth(ClientAuth.NONE)
-                .build());
-            return;
-        }
-
-        String tlsKeyPath = ConfigurationManager.getProxyConfig().getGrpcTlsKeyPath();
-        String tlsCertPath = ConfigurationManager.getProxyConfig().getGrpcTlsCertPath();
-        try (InputStream serverKeyInputStream = new FileInputStream(tlsKeyPath);
-             InputStream serverCertificateStream = new FileInputStream(tlsCertPath)) {
-            serverBuilder.sslContext(GrpcSslContexts.forServer(serverCertificateStream, serverKeyInputStream)
-                .trustManager(InsecureTrustManagerFactory.INSTANCE)
-                .clientAuth(ClientAuth.NONE)
-                .build());
-            log.info("TLS configured OK");
-        } catch (IOException e) {
-            log.error("Failed to load Server key/certificate", e);
-        }
-    }
-
-    protected void configInterceptor(NettyServerBuilder serverBuilder) {
-        // grpc interceptors, including acl, logging etc.
-        if (ConfigurationManager.getProxyConfig().isEnableACL()) {
-            List<AccessValidator> accessValidators = ServiceProvider.load(ServiceProvider.ACL_VALIDATOR_ID, AccessValidator.class);
-            if (accessValidators.isEmpty()) {
-                throw new IllegalArgumentException("Load AccessValidator failed");
-            }
-            serverBuilder.intercept(new AuthenticationInterceptor(accessValidators));
-        }
-
-        serverBuilder.intercept(new ContextInterceptor())
-            .intercept(new HeaderInterceptor());
+    protected GrpcServer(io.grpc.Server server) {
+        this.server = server;
     }
 
     public void start() throws Exception {
-        // first to start grpc service.
-        this.grpcForwardService.start();
-
         this.server.start();
         log.info("grpc server start successfully.");
     }
@@ -170,10 +40,6 @@ public class GrpcServer implements StartAndShutdown {
     public void shutdown() {
         try {
             this.server.shutdown().awaitTermination(30, TimeUnit.SECONDS);
-            this.executor.shutdown();
-
-            this.grpcForwardService.shutdown();
-
             log.info("grpc server shutdown successfully.");
         } catch (Exception e) {
             e.printStackTrace();

proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServerBuilder.java

@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.proxy.grpc;
+
+import io.grpc.BindableService;
+import io.grpc.ServerServiceDefinition;
+import io.grpc.netty.shaded.io.grpc.netty.GrpcSslContexts;
+import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder;
+import io.grpc.netty.shaded.io.netty.channel.epoll.EpollEventLoopGroup;
+import io.grpc.netty.shaded.io.netty.channel.epoll.EpollServerSocketChannel;
+import io.grpc.netty.shaded.io.netty.channel.nio.NioEventLoopGroup;
+import io.grpc.netty.shaded.io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.grpc.netty.shaded.io.netty.handler.ssl.ClientAuth;
+import io.grpc.netty.shaded.io.netty.handler.ssl.util.InsecureTrustManagerFactory;
+import io.grpc.netty.shaded.io.netty.handler.ssl.util.SelfSignedCertificate;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.CertificateException;
+import java.util.List;
+import java.util.concurrent.ThreadPoolExecutor;
+import javax.net.ssl.SSLException;
+import org.apache.rocketmq.acl.AccessValidator;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.common.utils.ServiceProvider;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
+import org.apache.rocketmq.proxy.config.ConfigurationManager;
+import org.apache.rocketmq.proxy.config.ProxyConfig;
+import org.apache.rocketmq.proxy.grpc.interceptor.AuthenticationInterceptor;
+import org.apache.rocketmq.proxy.grpc.interceptor.ContextInterceptor;
+import org.apache.rocketmq.proxy.grpc.interceptor.HeaderInterceptor;
+
+public class GrpcServerBuilder {
+    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
+    protected NettyServerBuilder serverBuilder;
+
+    public static GrpcServerBuilder newBuilder(ThreadPoolExecutor executor) {
+        return new GrpcServerBuilder(executor);
+    }
+
+    public GrpcServerBuilder(ThreadPoolExecutor executor) {
+        int port = ConfigurationManager.getProxyConfig().getGrpcServerPort();
+        serverBuilder = NettyServerBuilder.forPort(port);
+
+        try {
+            configSslContext(serverBuilder);
+        } catch (Exception e) {
+            log.error("grpc tls set failed. msg: {}, e:", e.getMessage(), e);
+            throw new RuntimeException("grpc tls set failed: " + e.getMessage());
+        }
+
+        // build server
+        int bossLoopNum = ConfigurationManager.getProxyConfig().getGrpcBossLoopNum();
+        int workerLoopNum = ConfigurationManager.getProxyConfig().getGrpcWorkerLoopNum();
+        int maxInboundMessageSize = ConfigurationManager.getProxyConfig().getGrpcMaxInboundMessageSize();
+
+        if (ConfigurationManager.getProxyConfig().isEnableGrpcEpoll()) {
+            serverBuilder.maxInboundMessageSize(maxInboundMessageSize)
+                .bossEventLoopGroup(new EpollEventLoopGroup(bossLoopNum))
+                .workerEventLoopGroup(new EpollEventLoopGroup(workerLoopNum))
+                .channelType(EpollServerSocketChannel.class)
+                .executor(executor);
+        } else {
+            serverBuilder.maxInboundMessageSize(maxInboundMessageSize)
+                .bossEventLoopGroup(new NioEventLoopGroup(bossLoopNum))
+                .workerEventLoopGroup(new NioEventLoopGroup(workerLoopNum))
+                .channelType(NioServerSocketChannel.class)
+                .executor(executor);
+        }
+
+        configInterceptor(serverBuilder);
+
+        log.info(
+            "grpc server has built. port: {}, tlsKeyPath: {}, tlsCertPath: {}, threadPool: {}, queueCapacity: {}, "
+                + "boosLoop: {}, workerLoop: {}, maxInboundMessageSize: {}",
+            port, bossLoopNum, workerLoopNum, maxInboundMessageSize);
+    }
+
+    public GrpcServerBuilder addService(BindableService service) {
+        this.serverBuilder.addService(service);
+        return this;
+    }
+
+    public GrpcServerBuilder addService(ServerServiceDefinition service) {
+        this.serverBuilder.addService(service);
+        return this;
+    }
+
+    public GrpcServer build() {
+        return new GrpcServer(this.serverBuilder.build());
+    }
+
+    protected void configSslContext(NettyServerBuilder serverBuilder) throws SSLException, CertificateException {
+        if (null == serverBuilder) {
+            return;
+        }
+        ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig();
+        boolean tlsTestModeEnable = proxyConfig.isGrpcTlsTestModeEnable();
+        if (tlsTestModeEnable) {
+            SelfSignedCertificate selfSignedCertificate = new SelfSignedCertificate();
+            serverBuilder.sslContext(GrpcSslContexts.forServer(selfSignedCertificate.certificate(), selfSignedCertificate.privateKey())
+                .trustManager(InsecureTrustManagerFactory.INSTANCE)
+                .clientAuth(ClientAuth.NONE)
+                .build());
+            return;
+        }
+
+        String tlsKeyPath = ConfigurationManager.getProxyConfig().getGrpcTlsKeyPath();
+        String tlsCertPath = ConfigurationManager.getProxyConfig().getGrpcTlsCertPath();
+        try (InputStream serverKeyInputStream = new FileInputStream(tlsKeyPath);
+             InputStream serverCertificateStream = new FileInputStream(tlsCertPath)) {
+            serverBuilder.sslContext(GrpcSslContexts.forServer(serverCertificateStream, serverKeyInputStream)
+                .trustManager(InsecureTrustManagerFactory.INSTANCE)
+                .clientAuth(ClientAuth.NONE)
+                .build());
+            log.info("TLS configured OK");
+        } catch (IOException e) {
+            log.error("Failed to load Server key/certificate", e);
+        }
+    }
+
+    protected void configInterceptor(NettyServerBuilder serverBuilder) {
+        // grpc interceptors, including acl, logging etc.
+        if (ConfigurationManager.getProxyConfig().isEnableACL()) {
+            List<AccessValidator> accessValidators = ServiceProvider.load(ServiceProvider.ACL_VALIDATOR_ID, AccessValidator.class);
+            if (accessValidators.isEmpty()) {
+                throw new IllegalArgumentException("Load AccessValidator failed");
+            }
+            serverBuilder.intercept(new AuthenticationInterceptor(accessValidators));
+        }
+
+        serverBuilder.intercept(new ContextInterceptor())
+            .intercept(new HeaderInterceptor());
+    }
+}