refactor(proxy): 重构 TLS证书管理

- 将 TlsCertificateManager 实例化移至 ProxyStartup 类
- 更新 GrpcServer 和 RemotingProtocolServer 类以使用 TlsCertificateManager
- 移除冗余的 TLS 证书管理相关测试用例
- 优化 TLS 上下文重载逻辑

Signed-off-by: Async <raisinata@foxmail.com>

Author: EnableAsync

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

@@ -82,7 +82,8 @@ public static void main(String[] args) {
             PROXY_START_AND_SHUTDOWN.appendStartAndShutdown(tlsCertificateManager);
 
             // create grpcServer
-            GrpcServer grpcServer = GrpcServerBuilder.newBuilder(executor, ConfigurationManager.getProxyConfig().getGrpcServerPort())
+            GrpcServer grpcServer = GrpcServerBuilder.newBuilder(executor,
+                    ConfigurationManager.getProxyConfig().getGrpcServerPort(), tlsCertificateManager)
                 .addService(createServiceProcessor(messagingProcessor))
                 .addService(ChannelzService.newInstance(100))
                 .addService(ProtoReflectionService.newInstance())
@@ -91,7 +92,7 @@ public static void main(String[] args) {
                 .build();
             PROXY_START_AND_SHUTDOWN.appendStartAndShutdown(grpcServer);
 
-            RemotingProtocolServer remotingServer = new RemotingProtocolServer(messagingProcessor);
+            RemotingProtocolServer remotingServer = new RemotingProtocolServer(messagingProcessor, tlsCertificateManager);
             PROXY_START_AND_SHUTDOWN.appendStartAndShutdown(remotingServer);
 
             // start servers one by one.

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

@@ -38,33 +38,35 @@ public class GrpcServer implements StartAndShutdown {
 
     private final TimeUnit unit;
 
-    final GrpcTlsReloadHandler tlsReloadHandler;
+    private final TlsCertificateManager tlsCertificateManager;
+    @VisibleForTesting final GrpcTlsReloadHandler tlsReloadHandler;
 
-    protected GrpcServer(Server server, long timeout, TimeUnit unit) throws Exception {
+    protected GrpcServer(Server server, long timeout, TimeUnit unit, TlsCertificateManager tlsCertificateManager) throws Exception {
         this.server = server;
         this.timeout = timeout;
         this.unit = unit;
-
+        this.tlsCertificateManager = tlsCertificateManager;
         this.tlsReloadHandler = new GrpcTlsReloadHandler();
-
-        // Register the TLS context reload handler
-        TlsCertificateManager.getInstance().registerReloadListener(this.tlsReloadHandler);
     }
 
     public void start() throws Exception {
+        // Register the TLS context reload handler
+        tlsCertificateManager.registerReloadListener(this.tlsReloadHandler);
+
         this.server.start();
         log.info("grpc server start successfully.");
     }
 
     public void shutdown() {
         try {
             // Unregister the TLS context reload handler
-            TlsCertificateManager.getInstance().unregisterReloadListener(this.tlsReloadHandler);
+            tlsCertificateManager.unregisterReloadListener(this.tlsReloadHandler);
 
             this.server.shutdown().awaitTermination(timeout, unit);
 
             log.info("grpc server shutdown successfully.");
         } catch (Exception e) {
+            e.printStackTrace();
             log.error("Failed to shutdown grpc server", e);
         }
     }

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

@@ -44,12 +44,14 @@ public class GrpcServerBuilder {
 
     protected TimeUnit unit = TimeUnit.SECONDS;
 
+    protected TlsCertificateManager tlsCertificateManager;
+
     public static GrpcServerBuilder newBuilder(ThreadPoolExecutor executor, int port, TlsCertificateManager tlsCertificateManager) {
         return new GrpcServerBuilder(executor, port, tlsCertificateManager);
     }
 
     protected GrpcServerBuilder(ThreadPoolExecutor executor, int port, TlsCertificateManager tlsCertificateManager) {
-//        tlsCertificateManager.registerReloadListener();
+        this.tlsCertificateManager = tlsCertificateManager;
         serverBuilder = NettyServerBuilder.forPort(port);
 
         serverBuilder.protocolNegotiator(new ProxyAndTlsProtocolNegotiator());
@@ -101,7 +103,7 @@ public GrpcServerBuilder appendInterceptor(ServerInterceptor interceptor) {
     }
 
     public GrpcServer build() throws Exception {
-        return new GrpcServer(this.serverBuilder.build(), time, unit);
+        return new GrpcServer(this.serverBuilder.build(), time, unit, tlsCertificateManager);
     }
 
     public GrpcServerBuilder configInterceptor() {

proxy/src/main/java/org/apache/rocketmq/proxy/remoting/MultiProtocolRemotingServer.java

@@ -68,7 +68,7 @@ public void loadSslContext() {
         if (tlsMode != TlsMode.DISABLED) {
             try {
                 sslContext = MultiProtocolTlsHelper.buildSslContext();
-                log.info("SSLContext created for remoting server");
+                log.info("SSLContext created for multi protocol remoting server");
             } catch (CertificateException | IOException e) {
                 throw new ProxyException(ProxyExceptionCode.INTERNAL_SERVER_ERROR, "Failed to create SSLContext for server", e);
             }

proxy/src/main/java/org/apache/rocketmq/proxy/remoting/MultiProtocolTlsHelper.java

@@ -61,12 +61,12 @@ public static SslContext buildSslContext() throws IOException, CertificateExcept
             log.info("Using JDK SSL provider");
         }
 
-        SslContextBuilder sslContextBuilder = null;
+        SslContextBuilder sslContextBuilder;
         if (tlsTestModeEnable) {
             SelfSignedCertificate selfSignedCertificate = new SelfSignedCertificate();
             sslContextBuilder = SslContextBuilder
                 .forServer(selfSignedCertificate.certificate(), selfSignedCertificate.privateKey())
-                .sslProvider(SslProvider.OPENSSL)
+                .sslProvider(provider)
                 .clientAuth(ClientAuth.OPTIONAL);
         } else {
             sslContextBuilder = SslContextBuilder.forServer(

proxy/src/main/java/org/apache/rocketmq/proxy/remoting/RemotingProtocolServer.java

@@ -89,10 +89,11 @@ public class RemotingProtocolServer implements StartAndShutdown, RemotingProxyOu
     protected final ThreadPoolExecutor topicRouteExecutor;
     protected final ThreadPoolExecutor defaultExecutor;
     protected final ScheduledExecutorService timerExecutor;
+    protected final TlsCertificateManager tlsCertificateManager;
     protected final RemotingTlsReloadHandler tlsReloadHandler;
 
 
-    public RemotingProtocolServer(MessagingProcessor messagingProcessor) throws Exception {
+    public RemotingProtocolServer(MessagingProcessor messagingProcessor, TlsCertificateManager tlsCertificateManager) throws Exception {
         this.messagingProcessor = messagingProcessor;
         this.remotingChannelManager = new RemotingChannelManager(this, messagingProcessor.getProxyRelayService());
 
@@ -117,6 +118,8 @@ public RemotingProtocolServer(MessagingProcessor messagingProcessor) throws Exce
         System.setProperty(TlsSystemConfig.TLS_SERVER_CERTPATH, config.getTlsCertPath());
         TlsSystemConfig.tlsServerKeyPath = config.getTlsKeyPath();
         System.setProperty(TlsSystemConfig.TLS_SERVER_KEYPATH, config.getTlsKeyPath());
+        this.tlsCertificateManager = tlsCertificateManager;
+        this.tlsReloadHandler = new RemotingTlsReloadHandler();
 
         this.clientHousekeepingService = new ClientHousekeepingService(this.clientManagerActivity);
 
@@ -191,9 +194,6 @@ public RemotingProtocolServer(MessagingProcessor messagingProcessor) throws Exce
         );
         this.timerExecutor.scheduleAtFixedRate(this::cleanExpireRequest, 10, 10, TimeUnit.SECONDS);
 
-        this.tlsReloadHandler = new RemotingTlsReloadHandler();
-        TlsCertificateManager.getInstance().registerReloadListener(this.tlsReloadHandler);
-
         this.registerRemotingServer(this.defaultRemotingServer);
     }
 
@@ -243,7 +243,7 @@ protected void registerRemotingServer(RemotingServer remotingServer) {
     @Override
     public void shutdown() throws Exception {
         // Unregister the TLS context reload handler
-        TlsCertificateManager.getInstance().unregisterReloadListener(this.tlsReloadHandler);
+        tlsCertificateManager.unregisterReloadListener(this.tlsReloadHandler);
 
         this.defaultRemotingServer.shutdown();
         this.remotingChannelManager.shutdown();
@@ -257,6 +257,9 @@ public void shutdown() throws Exception {
 
     @Override
     public void start() throws Exception {
+        // Register the TLS context reload handler
+        tlsCertificateManager.registerReloadListener(this.tlsReloadHandler);
+
         this.remotingChannelManager.start();
         this.defaultRemotingServer.start();
     }