[ISSUE #8920] Refactor SSL context loading process to support multiple protocols dynamic loading (#9483)

* feat(proxy): 添加 gRPC 和 Remoting 服务器的 TLS 证书热更新支持 - 在 GrpcServer 和 RemotingProtocolServer 中添加文件监视服务，用于监控 TLS 证书和密钥的变化 - 实现证书和密钥变更时重新加载 SSL 上下文的逻辑 - 优化 ProxyAndTlsProtocolNegotiator 中的 SSL 上下文加载过程 - 添加日志记录，方便调试和监控 TLS 相关操作 * refactor(proxy): 重构 gRPC 证书监控逻辑并添加单元测试 - 重构 GrpcServer 中的证书监控逻辑，提取到独立的 GrpcCertKeyFileWatchListener 类中 - 优化证书变更处理流程，提高代码可读性和维护性 - 新增 GrpcServerTest 类，为 gRPC服务器和证书监控添加单元测试- 测试覆盖了各种证书变更场景，包括单独变更和组合变更 - 验证了证书变更时 SSLContext 的重新加载和错误处理 Signed-off-by: Async <raisinata@foxmail.com> * refactor(proxy): 重构 gRPC 证书监控逻辑并添加单元测试 - 重构 GrpcServer 中的证书监控逻辑，提取到独立的 GrpcCertKeyFileWatchListener 类中 - 优化证书变更处理流程，提高代码可读性和维护性 - 新增 GrpcServerTest 类，为 gRPC服务器和证书监控添加单元测试- 测试覆盖了各种证书变更场景，包括单独变更和组合变更 - 验证了证书变更时 SSLContext 的重新加载和错误处理 Signed-off-by: Async <raisinata@foxmail.com> * fix: code format Signed-off-by: Async <raisinata@foxmail.com> * test: add test cases Signed-off-by: Async <raisinata@foxmail.com> * fix: code format Signed-off-by: Async <raisinata@foxmail.com> * refactor(proxy): 重构 TLS证书更新逻辑 - 移除 FileWatchService，改用 TlsCertificateManager 统一管理 TLS证书 - 实现 TlsContextReloadListener 接口，响应 TLS 证书更新 - 优化 GrpcServer 和 RemotingProtocolServer 中的 TLS 证书更新逻辑 - 新增单元测试验证 TLS 证书更新功能 Signed-off-by: Async <raisinata@foxmail.com> * test(proxy): 优化 TLS 相关测试用例 - 重构了多个测试类中的重复代码- 提高了测试的可读性和维护性 - 确保在测试中正确关闭资源 Signed-off-by: Async <raisinata@foxmail.com> * refactor(proxy): 优化代码导入结构 - 移除了不必要的导入项 - 显式导入了所有活动类，提高了代码的可读性和维护性 Signed-off-by: Async <raisinata@foxmail.com> * update * fix: no static Signed-off-by: Async <raisinata@foxmail.com> * fix: add SingletonHolder for TlsCertificateManager Signed-off-by: Async <raisinata@foxmail.com> * refactor * refactor(proxy): 重构 TLS证书管理 - 将 TlsCertificateManager 实例化移至 ProxyStartup 类 - 更新 GrpcServer 和 RemotingProtocolServer 类以使用 TlsCertificateManager - 移除冗余的 TLS 证书管理相关测试用例 - 优化 TLS 上下文重载逻辑 Signed-off-by: Async <raisinata@foxmail.com> * refactor(proxy): 优化日志信息内容 - 将 cert file changed 日志信息改为更通用的 File changed - 保持代码风格一致性，提高日志的可读性和维护性 Signed-off-by: Async <raisinata@foxmail.com> * test(proxy): 重构并增强 TlsCertificateManager 测试用例- 重新设计测试用例，使用临时文件模拟证书和密钥 - 增加对 TlsCertificateManager 各种方法的单元测试 - 涉及到的测试场景包括： - 构造函数 - 启动和关闭 - 注册和注销监听器 - 文件变更通知（证书、密钥、未知文件等） - 多个监听器的情况 - 监听器抛出异常的情况 - 增加对内部 CertKeyFileWatchListener 的测试 Signed-off-by: Async <raisinata@foxmail.com> * refactor * test(proxy): 优化 TlsCertificateManager 单元测试 -移除了未使用的 import 语句 - 替换了 import static语句，使其更加有序 - 删除了未使用的静态方法断言（verify、times、never） - 重置了 mock 对象以避免测试之间的干扰 Signed-off-by: Async <raisinata@foxmail.com> * fix format Signed-off-by: Async <raisinata@foxmail.com> * fix format Signed-off-by: Async <raisinata@foxmail.com> * fix format Signed-off-by: Async <raisinata@foxmail.com> * fix format Signed-off-by: Async <raisinata@foxmail.com> * fix format Signed-off-by: Async <raisinata@foxmail.com> * fix format Signed-off-by: Async <raisinata@foxmail.com> * fix format Signed-off-by: Async <raisinata@foxmail.com> * fix format Signed-off-by: Async <raisinata@foxmail.com> --------- Signed-off-by: Async <raisinata@foxmail.com>
2026-09-01 15:47:01 +08:00 · 2025-06-27 09:56:19 +08:00
parent d556460d86
commit 832562f028
13 changed files with 561 additions and 68 deletions
@@ -44,6 +44,7 @@ import org.apache.rocketmq.proxy.metrics.ProxyMetricsManager;
 import org.apache.rocketmq.proxy.processor.DefaultMessagingProcessor;
 import org.apache.rocketmq.proxy.processor.MessagingProcessor;
 import org.apache.rocketmq.proxy.remoting.RemotingProtocolServer;
+import org.apache.rocketmq.proxy.service.cert.TlsCertificateManager;
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 import org.apache.rocketmq.srvutil.ServerUtil;

@@ -76,8 +77,13 @@ public class ProxyStartup {

            MessagingProcessor messagingProcessor = createMessagingProcessor();

+            // tls cert update
+            TlsCertificateManager tlsCertificateManager = new TlsCertificateManager();
+            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())
@@ -86,7 +92,7 @@ public class ProxyStartup {
                .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.
@@ -17,12 +17,17 @@

 package org.apache.rocketmq.proxy.grpc;

-import java.util.concurrent.TimeUnit;
+import com.google.common.annotations.VisibleForTesting;
 import io.grpc.Server;
 import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.common.utils.StartAndShutdown;
 import org.apache.rocketmq.logging.org.slf4j.Logger;
 import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
-import org.apache.rocketmq.common.utils.StartAndShutdown;
+import org.apache.rocketmq.proxy.service.cert.TlsCertificateManager;
+
+import java.io.IOException;
+import java.security.cert.CertificateException;
+import java.util.concurrent.TimeUnit;

 public class GrpcServer implements StartAndShutdown {
    private static final Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
@@ -33,23 +38,50 @@ public class GrpcServer implements StartAndShutdown {

    private final TimeUnit unit;

-    protected GrpcServer(Server server, long timeout, TimeUnit unit) {
+    private final TlsCertificateManager tlsCertificateManager;
+    @VisibleForTesting final GrpcTlsReloadHandler tlsReloadHandler;
+
+    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();
    }

    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.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);
        }
    }
-}
+
+    @VisibleForTesting
+    class GrpcTlsReloadHandler implements TlsCertificateManager.TlsContextReloadListener {
+        @Override
+        public void onTlsContextReload() {
+            try {
+                ProxyAndTlsProtocolNegotiator.loadSslContext();
+                log.info("SslContext reloaded for grpc server");
+            } catch (CertificateException | IOException e) {
+                log.error("Failed to reload SslContext for server", e);
+            }
+        }
+    }
+}
@@ -34,6 +34,7 @@ import org.apache.rocketmq.proxy.grpc.interceptor.HeaderInterceptor;

 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import org.apache.rocketmq.proxy.service.cert.TlsCertificateManager;

 public class GrpcServerBuilder {
    private static final Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
@@ -43,11 +44,15 @@ public class GrpcServerBuilder {

    protected TimeUnit unit = TimeUnit.SECONDS;

-    public static GrpcServerBuilder newBuilder(ThreadPoolExecutor executor, int port) {
-        return new GrpcServerBuilder(executor, port);
+    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) {
+    protected GrpcServerBuilder(ThreadPoolExecutor executor, int port, TlsCertificateManager tlsCertificateManager) {
+        this.tlsCertificateManager = tlsCertificateManager;
        serverBuilder = NettyServerBuilder.forPort(port);

        serverBuilder.protocolNegotiator(new ProxyAndTlsProtocolNegotiator());
@@ -71,7 +76,7 @@ public class GrpcServerBuilder {
        }

        serverBuilder.maxInboundMessageSize(maxInboundMessageSize)
-                .maxConnectionIdle(idleTimeMills, TimeUnit.MILLISECONDS);
+            .maxConnectionIdle(idleTimeMills, TimeUnit.MILLISECONDS);

        log.info("grpc server has built. port: {}, bossLoopNum: {}, workerLoopNum: {}, maxInboundMessageSize: {}",
            port, bossLoopNum, workerLoopNum, maxInboundMessageSize);
@@ -98,8 +103,8 @@ public class GrpcServerBuilder {
        return this;
    }

-    public GrpcServer build() {
-        return new GrpcServer(this.serverBuilder.build(), time, unit);
+    public GrpcServer build() throws Exception {
+        return new GrpcServer(this.serverBuilder.build(), time, unit, tlsCertificateManager);
    }

    public GrpcServerBuilder configInterceptor() {
@@ -36,16 +36,23 @@ import io.grpc.netty.shaded.io.netty.handler.codec.haproxy.HAProxyMessageDecoder
 import io.grpc.netty.shaded.io.netty.handler.codec.haproxy.HAProxyProtocolVersion;
 import io.grpc.netty.shaded.io.netty.handler.codec.haproxy.HAProxyTLV;
 import io.grpc.netty.shaded.io.netty.handler.ssl.ClientAuth;
+import io.grpc.netty.shaded.io.netty.handler.ssl.OpenSsl;
 import io.grpc.netty.shaded.io.netty.handler.ssl.SslContext;
 import io.grpc.netty.shaded.io.netty.handler.ssl.SslHandler;
+import io.grpc.netty.shaded.io.netty.handler.ssl.SslProvider;
+
 import io.grpc.netty.shaded.io.netty.handler.ssl.util.InsecureTrustManagerFactory;
 import io.grpc.netty.shaded.io.netty.handler.ssl.util.SelfSignedCertificate;
 import io.grpc.netty.shaded.io.netty.util.AsciiString;
 import io.grpc.netty.shaded.io.netty.util.CharsetUtil;
+
+import java.io.IOException;
 import java.io.InputStream;
 import java.nio.file.Files;
 import java.nio.file.Paths;
+import java.security.cert.CertificateException;
 import java.util.List;
+
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.rocketmq.common.constant.HAProxyConstants;
@@ -73,7 +80,13 @@ public class ProxyAndTlsProtocolNegotiator implements InternalProtocolNegotiator
    private static SslContext sslContext;

    public ProxyAndTlsProtocolNegotiator() {
-        sslContext = loadSslContext();
+        try {
+            loadSslContext();
+            log.info("SslContext created for proxy server");
+        } catch (IOException | CertificateException e) {
+            log.error("SslContext init error", e);
+            throw new RuntimeException(e);
+        }
    }

    @Override
@@ -90,35 +103,36 @@ public class ProxyAndTlsProtocolNegotiator implements InternalProtocolNegotiator
    public void close() {
    }

-    private static SslContext loadSslContext() {
-        try {
-            ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig();
-            if (proxyConfig.isTlsTestModeEnable()) {
-                SelfSignedCertificate selfSignedCertificate = new SelfSignedCertificate();
-                return GrpcSslContexts.forServer(selfSignedCertificate.certificate(),
-                                selfSignedCertificate.privateKey())
-                        .trustManager(InsecureTrustManagerFactory.INSTANCE)
-                        .clientAuth(ClientAuth.NONE)
-                        .build();
-            } else {
-                String tlsKeyPath = ConfigurationManager.getProxyConfig().getTlsKeyPath();
-                String tlsCertPath = ConfigurationManager.getProxyConfig().getTlsCertPath();
-                try (InputStream serverKeyInputStream = Files.newInputStream(
-                        Paths.get(tlsKeyPath));
-                     InputStream serverCertificateStream = Files.newInputStream(
-                             Paths.get(tlsCertPath))) {
-                    SslContext res = GrpcSslContexts.forServer(serverCertificateStream,
-                                    serverKeyInputStream)
-                            .trustManager(InsecureTrustManagerFactory.INSTANCE)
-                            .clientAuth(ClientAuth.NONE)
-                            .build();
-                    log.info("grpc load TLS configured OK");
-                    return res;
-                }
+    public static void loadSslContext() throws CertificateException, IOException {
+        ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig();
+        SslProvider provider;
+        if (OpenSsl.isAvailable()) {
+            provider = SslProvider.OPENSSL;
+            log.info("Using OpenSSL provider");
+        } else {
+            provider = SslProvider.JDK;
+            log.info("Using JDK SSL provider");
+        }
+        if (proxyConfig.isTlsTestModeEnable()) {
+            SelfSignedCertificate selfSignedCertificate = new SelfSignedCertificate();
+            sslContext = GrpcSslContexts.forServer(selfSignedCertificate.certificate(), selfSignedCertificate.privateKey())
+                .sslProvider(provider)
+                .trustManager(InsecureTrustManagerFactory.INSTANCE)
+                .clientAuth(ClientAuth.NONE)
+                .build();
+        } else {
+            String tlsCertPath = ConfigurationManager.getProxyConfig().getTlsCertPath();
+            String tlsKeyPath = ConfigurationManager.getProxyConfig().getTlsKeyPath();
+            try (InputStream serverKeyInputStream = Files.newInputStream(
+                Paths.get(tlsKeyPath));
+                 InputStream serverCertificateStream = Files.newInputStream(
+                     Paths.get(tlsCertPath))) {
+                sslContext = GrpcSslContexts.forServer(serverCertificateStream,
+                        serverKeyInputStream)
+                    .trustManager(InsecureTrustManagerFactory.INSTANCE)
+                    .clientAuth(ClientAuth.NONE)
+                    .build();
            }
-        } catch (Exception e) {
-            log.error("grpc tls set failed. msg: {}, e:", e.getMessage(), e);
-            throw new RuntimeException("grpc tls set failed: " + e.getMessage());
        }
    }

@@ -135,15 +149,14 @@ public class ProxyAndTlsProtocolNegotiator implements InternalProtocolNegotiator
        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
            try {
-                ProtocolDetectionResult<HAProxyProtocolVersion> ha = HAProxyMessageDecoder.detectProtocol(
-                        in);
+                ProtocolDetectionResult<HAProxyProtocolVersion> ha = HAProxyMessageDecoder.detectProtocol(in);
                if (ha.state() == ProtocolDetectionState.NEEDS_MORE_DATA) {
                    return;
                }
                if (ha.state() == ProtocolDetectionState.DETECTED) {
                    ctx.pipeline().addAfter(ctx.name(), HA_PROXY_DECODER, new HAProxyMessageDecoder())
-                            .addAfter(HA_PROXY_DECODER, HA_PROXY_HANDLER, new HAProxyMessageHandler())
-                            .addAfter(HA_PROXY_HANDLER, TLS_MODE_HANDLER, new TlsModeHandler(grpcHandler));
+                        .addAfter(HA_PROXY_DECODER, HA_PROXY_HANDLER, new HAProxyMessageHandler())
+                        .addAfter(HA_PROXY_HANDLER, TLS_MODE_HANDLER, new TlsModeHandler(grpcHandler));
                } else {
                    ctx.pipeline().addAfter(ctx.name(), TLS_MODE_HANDLER, new TlsModeHandler(grpcHandler));
                }
@@ -209,7 +222,7 @@ public class ProxyAndTlsProtocolNegotiator implements InternalProtocolNegotiator
                    msg.tlvs().forEach(tlv -> handleHAProxyTLV(tlv, builder));
                }
                pne = InternalProtocolNegotiationEvent
-                        .withAttributes(InternalProtocolNegotiationEvent.getDefault(), builder.build());
+                    .withAttributes(InternalProtocolNegotiationEvent.getDefault(), builder.build());
            } finally {
                msg.release();
            }
@@ -244,9 +257,9 @@ public class ProxyAndTlsProtocolNegotiator implements InternalProtocolNegotiator

        public TlsModeHandler(GrpcHttp2ConnectionHandler grpcHandler) {
            this.ssl = InternalProtocolNegotiators.serverTls(sslContext)
-                    .newHandler(grpcHandler);
+                .newHandler(grpcHandler);
            this.plaintext = InternalProtocolNegotiators.serverPlaintext()
-                    .newHandler(grpcHandler);
+                .newHandler(grpcHandler);
        }

        @Override
@@ -258,7 +271,7 @@ public class ProxyAndTlsProtocolNegotiator implements InternalProtocolNegotiator
                } else if (TlsMode.DISABLED.equals(tlsMode)) {
                    ctx.pipeline().addAfter(ctx.name(), null, this.plaintext);
                } else {
-                    // in SslHandler.isEncrypted, it need at least 5 bytes to judge is encrypted or not
+                    // in SslHandler.isEncrypted, it needs at least 5 bytes to judge is encrypted or not
                    if (in.readableBytes() < SSL_RECORD_HEADER_LENGTH) {
                        return;
                    }
@@ -68,9 +68,9 @@ public class MultiProtocolRemotingServer extends NettyRemotingServer {
        if (tlsMode != TlsMode.DISABLED) {
            try {
                sslContext = MultiProtocolTlsHelper.buildSslContext();
-                log.info("SSLContext created for 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);
+                throw new ProxyException(ProxyExceptionCode.INTERNAL_SERVER_ERROR, "Failed to create SslContext for server", e);
            }
        }
    }
@@ -61,12 +61,12 @@ public class MultiProtocolTlsHelper extends TlsHelper {
            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(
@@ -46,6 +46,7 @@ import org.apache.rocketmq.proxy.remoting.pipeline.AuthenticationPipeline;
 import org.apache.rocketmq.proxy.remoting.pipeline.AuthorizationPipeline;
 import org.apache.rocketmq.proxy.remoting.pipeline.ContextInitPipeline;
 import org.apache.rocketmq.proxy.remoting.pipeline.RequestPipeline;
+import org.apache.rocketmq.proxy.service.cert.TlsCertificateManager;
 import org.apache.rocketmq.remoting.ChannelEventListener;
 import org.apache.rocketmq.remoting.InvokeCallback;
 import org.apache.rocketmq.remoting.RemotingServer;
@@ -88,8 +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) {
+
+    public RemotingProtocolServer(MessagingProcessor messagingProcessor, TlsCertificateManager tlsCertificateManager) throws Exception {
        this.messagingProcessor = messagingProcessor;
        this.remotingChannelManager = new RemotingChannelManager(this, messagingProcessor.getProxyRelayService());

@@ -114,6 +118,8 @@ public class RemotingProtocolServer implements StartAndShutdown, RemotingProxyOu
        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,6 +197,16 @@ public class RemotingProtocolServer implements StartAndShutdown, RemotingProxyOu
        this.registerRemotingServer(this.defaultRemotingServer);
    }

+    protected class RemotingTlsReloadHandler implements TlsCertificateManager.TlsContextReloadListener {
+        @Override
+        public void onTlsContextReload() {
+            if (defaultRemotingServer instanceof NettyRemotingServer) {
+                ((NettyRemotingServer) defaultRemotingServer).loadSslContext();
+                log.info("SslContext reloaded for remoting server");
+            }
+        }
+    }
+
    protected void registerRemotingServer(RemotingServer remotingServer) {
        remotingServer.registerProcessor(RequestCode.SEND_MESSAGE, sendMessageActivity, this.sendMessageExecutor);
        remotingServer.registerProcessor(RequestCode.SEND_MESSAGE_V2, sendMessageActivity, this.sendMessageExecutor);
@@ -226,6 +242,9 @@ public class RemotingProtocolServer implements StartAndShutdown, RemotingProxyOu

    @Override
    public void shutdown() throws Exception {
+        // Unregister the TLS context reload handler
+        tlsCertificateManager.unregisterReloadListener(this.tlsReloadHandler);
+
        this.defaultRemotingServer.shutdown();
        this.remotingChannelManager.shutdown();
        this.sendMessageExecutor.shutdown();
@@ -238,6 +257,9 @@ public class RemotingProtocolServer implements StartAndShutdown, RemotingProxyOu

    @Override
    public void start() throws Exception {
+        // Register the TLS context reload handler
+        tlsCertificateManager.registerReloadListener(this.tlsReloadHandler);
+
        this.remotingChannelManager.start();
        this.defaultRemotingServer.start();
    }
@@ -75,8 +75,8 @@ public class Http2ProtocolProxyHandler implements ProtocolHandler {
                    .build();
            }
        } catch (SSLException e) {
-            log.error("Failed to create SSLContext for Http2ProtocolProxyHandler", e);
-            throw new RuntimeException("Failed to create SSLContext for Http2ProtocolProxyHandler", e);
+            log.error("Failed to create SslContext for Http2ProtocolProxyHandler", e);
+            throw new RuntimeException("Failed to create SslContext for Http2ProtocolProxyHandler", e);
        }
    }

@@ -0,0 +1,121 @@
+/*
+ * 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.service.cert;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.common.utils.StartAndShutdown;
+import org.apache.rocketmq.logging.org.slf4j.Logger;
+import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
+import org.apache.rocketmq.proxy.config.ConfigurationManager;
+import org.apache.rocketmq.remoting.netty.TlsSystemConfig;
+import org.apache.rocketmq.srvutil.FileWatchService;
+import java.util.ArrayList;
+import java.util.List;
+
+public class TlsCertificateManager implements StartAndShutdown {
+    private static final Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
+
+    private final FileWatchService fileWatchService;
+    private final List<TlsContextReloadListener> reloadListeners = new ArrayList<>();
+
+    public TlsCertificateManager() {
+        try {
+            this.fileWatchService = new FileWatchService(
+                new String[] {
+                    ConfigurationManager.getProxyConfig().getTlsCertPath(),
+                    ConfigurationManager.getProxyConfig().getTlsKeyPath()
+                },
+                new CertKeyFileWatchListener(),
+                60 * 60 * 1000 /* 1 hour */
+            );
+        } catch (Exception e) {
+            log.error("Failed to initialize TLS certificate watch service", e);
+            throw new RuntimeException("Failed to initialize TLS certificate manager", e);
+        }
+    }
+
+    public FileWatchService getFileWatchService() {
+        return this.fileWatchService;
+    }
+
+    public void registerReloadListener(TlsContextReloadListener listener) {
+        if (listener != null) {
+            this.reloadListeners.add(listener);
+        }
+    }
+
+    public void unregisterReloadListener(TlsContextReloadListener listener) {
+        if (listener != null) {
+            this.reloadListeners.remove(listener);
+        }
+    }
+
+    public List<TlsContextReloadListener> getReloadListeners() {
+        return this.reloadListeners;
+    }
+
+    @Override
+    public void start() throws Exception {
+        this.fileWatchService.start();
+        log.info("TLS certificate manager started successfully, start watching: {} {}",
+            ConfigurationManager.getProxyConfig().getTlsCertPath(),
+            ConfigurationManager.getProxyConfig().getTlsKeyPath()
+        );
+    }
+
+    @Override
+    public void shutdown() throws Exception {
+        this.fileWatchService.shutdown();
+        log.info("TLS certificate manager shutdown successfully");
+    }
+
+    private class CertKeyFileWatchListener implements FileWatchService.Listener {
+        private boolean certChanged = false;
+        private boolean keyChanged = false;
+
+        @Override
+        public void onChanged(String path) {
+            log.info("File changed: {}", path);
+            if (path.equals(TlsSystemConfig.tlsServerCertPath)) {
+                certChanged = true;
+            } else if (path.equals(TlsSystemConfig.tlsServerKeyPath)) {
+                keyChanged = true;
+            }
+
+            if (certChanged && keyChanged) {
+                log.info("The certificate and private key changed, reload the ssl context");
+                notifyContextReload();
+                certChanged = false;
+                keyChanged = false;
+            }
+        }
+
+        private void notifyContextReload() {
+            for (TlsContextReloadListener listener : reloadListeners) {
+                try {
+                    listener.onTlsContextReload();
+                } catch (Exception e) {
+                    log.error("Failed to notify TLS context reload to listener: " + listener, e);
+                }
+            }
+        }
+    }
+
+    // Interface for listeners interested in TLS context reload events
+    public interface TlsContextReloadListener {
+        void onTlsContextReload();
+    }
+}
@@ -0,0 +1,289 @@
+/*
+ * 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.service.cert;
+
+import java.io.FileWriter;
+import org.apache.rocketmq.proxy.config.ConfigurationManager;
+import org.apache.rocketmq.proxy.config.ProxyConfig;
+import org.apache.rocketmq.remoting.netty.TlsSystemConfig;
+import org.apache.rocketmq.srvutil.FileWatchService;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.io.File;
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.nio.file.Path;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+@ExtendWith(MockitoExtension.class)
+public class TlsCertificateManagerTest {
+
+    @TempDir
+    Path tempDir;
+
+    private TlsCertificateManager manager;
+
+    @Mock
+    private ProxyConfig proxyConfig;
+
+    @Mock
+    private TlsCertificateManager.TlsContextReloadListener listener1;
+
+    @Mock
+    private TlsCertificateManager.TlsContextReloadListener listener2;
+
+    private File certFile;
+    private File keyFile;
+    private FileWatchService.Listener fileWatchListener;
+    private Field configField;
+    private ProxyConfig originalConfig;
+
+    @BeforeAll
+    public static void setUpAll() throws Exception {
+        ConfigurationManager.initEnv();
+        ConfigurationManager.intConfig();
+    }
+
+    @BeforeEach
+    public void setUp() throws Exception {
+        // Create temporary certificate and key files
+        certFile = new File(tempDir.toFile(), "server.crt");
+        keyFile = new File(tempDir.toFile(), "server.key");
+        try (FileWriter certWriter = new FileWriter(certFile);
+             FileWriter keyWriter = new FileWriter(keyFile)) {
+            certWriter.write("test certificate content");
+            keyWriter.write("test key content");
+        }
+
+        // Set TlsSystemConfig paths
+        TlsSystemConfig.tlsServerCertPath = certFile.getAbsolutePath();
+        TlsSystemConfig.tlsServerKeyPath = keyFile.getAbsolutePath();
+
+        // Create the TlsCertificateManager
+        manager = new TlsCertificateManager();
+
+        // Extract the file watch listener using reflection
+        fileWatchListener = extractFileWatchListener(manager);
+    }
+
+    @AfterEach
+    public void tearDown() throws Exception {
+        // Restore the original config
+        if (configField != null && originalConfig != null) {
+            configField.set(null, originalConfig);
+        }
+    }
+
+    private FileWatchService.Listener extractFileWatchListener(TlsCertificateManager manager) throws Exception {
+        Field fileWatchServiceField = TlsCertificateManager.class.getDeclaredField("fileWatchService");
+        fileWatchServiceField.setAccessible(true);
+        FileWatchService fileWatchService = (FileWatchService) fileWatchServiceField.get(manager);
+
+        Field listenerField = FileWatchService.class.getDeclaredField("listener");
+        listenerField.setAccessible(true);
+        return (FileWatchService.Listener) listenerField.get(fileWatchService);
+    }
+
+    @Test
+    public void testConstructor() {
+        // The constructor should initialize the FileWatchService with the correct paths
+        assertNotNull(manager);
+    }
+
+    @Test
+    public void testStartAndShutdown() throws Exception {
+        TlsCertificateManager managerSpy = spy(manager);
+
+        Field watchServiceField = TlsCertificateManager.class.getDeclaredField("fileWatchService");
+        watchServiceField.setAccessible(true);
+        FileWatchService watchService = (FileWatchService) watchServiceField.get(managerSpy);
+        FileWatchService watchServiceSpy = spy(watchService);
+        watchServiceField.set(managerSpy, watchServiceSpy);
+
+        managerSpy.start();
+        verify(watchServiceSpy).start();
+
+        managerSpy.shutdown();
+        verify(watchServiceSpy).shutdown();
+    }
+
+    @Test
+    public void testRegisterAndUnregisterListener() {
+        manager.registerReloadListener(listener1);
+
+        List<TlsCertificateManager.TlsContextReloadListener> listeners = manager.getReloadListeners();
+        assertEquals(1, listeners.size());
+        assertTrue(listeners.contains(listener1));
+
+        manager.registerReloadListener(listener2);
+        assertEquals(2, listeners.size());
+        assertTrue(listeners.contains(listener2));
+
+        manager.unregisterReloadListener(listener1);
+        assertEquals(1, listeners.size());
+        assertFalse(listeners.contains(listener1));
+        assertTrue(listeners.contains(listener2));
+
+        manager.registerReloadListener(null);
+        assertEquals(1, listeners.size()); // Should remain unchanged
+
+        manager.unregisterReloadListener(null);
+        assertEquals(1, listeners.size()); // Should remain unchanged
+    }
+
+    @Test
+    public void testFileChangeNotification_CertOnly() throws Exception {
+        manager.registerReloadListener(listener1);
+
+        fileWatchListener.onChanged(certFile.getAbsolutePath());
+
+        verify(listener1, never()).onTlsContextReload();
+    }
+
+    @Test
+    public void testFileChangeNotification_KeyOnly() throws Exception {
+        manager.registerReloadListener(listener1);
+
+        fileWatchListener.onChanged(keyFile.getAbsolutePath());
+
+        verify(listener1, never()).onTlsContextReload();
+    }
+
+    @Test
+    public void testFileChangeNotification_BothFiles() throws Exception {
+        manager.registerReloadListener(listener1);
+
+        fileWatchListener.onChanged(certFile.getAbsolutePath());
+        fileWatchListener.onChanged(keyFile.getAbsolutePath());
+
+        verify(listener1, times(1)).onTlsContextReload();
+    }
+
+    @Test
+    public void testFileChangeNotification_MultipleListeners() throws Exception {
+        manager.registerReloadListener(listener1);
+        manager.registerReloadListener(listener2);
+
+        fileWatchListener.onChanged(certFile.getAbsolutePath());
+        fileWatchListener.onChanged(keyFile.getAbsolutePath());
+
+        verify(listener1, times(1)).onTlsContextReload();
+        verify(listener2, times(1)).onTlsContextReload();
+    }
+
+    @Test
+    public void testFileChangeNotification_BothFilesReverseOrder() throws Exception {
+        manager.registerReloadListener(listener1);
+
+        fileWatchListener.onChanged(keyFile.getAbsolutePath());
+        fileWatchListener.onChanged(certFile.getAbsolutePath());
+
+        verify(listener1, times(1)).onTlsContextReload();
+    }
+
+    @Test
+    public void testFileChangeNotification_RepeatedChanges() throws Exception {
+        manager.registerReloadListener(listener1);
+
+        fileWatchListener.onChanged(certFile.getAbsolutePath());
+        fileWatchListener.onChanged(keyFile.getAbsolutePath());
+
+        verify(listener1, times(1)).onTlsContextReload();
+
+        fileWatchListener.onChanged(certFile.getAbsolutePath());
+        fileWatchListener.onChanged(keyFile.getAbsolutePath());
+
+        verify(listener1, times(2)).onTlsContextReload();
+    }
+
+    @Test
+    public void testFileChangeNotification_UnknownFile() throws Exception {
+        manager.registerReloadListener(listener1);
+
+        fileWatchListener.onChanged("/unknown/file/path");
+
+        verify(listener1, never()).onTlsContextReload();
+    }
+
+    @Test
+    public void testFileChangeNotification_ListenerThrowsException() throws Exception {
+        TlsCertificateManager.TlsContextReloadListener exceptionListener = mock(TlsCertificateManager.TlsContextReloadListener.class);
+        doThrow(new RuntimeException("Test exception")).when(exceptionListener).onTlsContextReload();
+
+        manager.registerReloadListener(exceptionListener);
+        manager.registerReloadListener(listener1);
+
+        fileWatchListener.onChanged(certFile.getAbsolutePath());
+        fileWatchListener.onChanged(keyFile.getAbsolutePath());
+
+        verify(exceptionListener, times(1)).onTlsContextReload();
+        verify(listener1, times(1)).onTlsContextReload();
+    }
+
+    @Test
+    public void testInnerCertKeyFileWatchListener() throws Exception {
+        Class<?> innerClass = null;
+        for (Class<?> clazz : TlsCertificateManager.class.getDeclaredClasses()) {
+            if (clazz.getSimpleName().equals("CertKeyFileWatchListener")) {
+                innerClass = clazz;
+                break;
+            }
+        }
+
+        assertNotNull(innerClass, "CertKeyFileWatchListener class not found");
+
+        Constructor<?> constructor = innerClass.getDeclaredConstructor(TlsCertificateManager.class);
+        constructor.setAccessible(true);
+        Object innerListener = constructor.newInstance(manager);
+
+        manager.registerReloadListener(listener1);
+
+        Method onChangedMethod = innerClass.getDeclaredMethod("onChanged", String.class);
+        onChangedMethod.setAccessible(true);
+
+        onChangedMethod.invoke(innerListener, certFile.getAbsolutePath());
+        verify(listener1, never()).onTlsContextReload();
+
+        onChangedMethod.invoke(innerListener, keyFile.getAbsolutePath());
+        verify(listener1, times(1)).onTlsContextReload();
+
+        reset(listener1);
+
+        onChangedMethod.invoke(innerListener, certFile.getAbsolutePath());
+        verify(listener1, never()).onTlsContextReload();
+    }
+}
@@ -167,10 +167,10 @@ public class NettyRemotingClient extends NettyRemotingAbstract implements Remoti
                sslContext = TlsHelper.buildSslContext(true);
                LOGGER.info("SSL enabled for client");
            } catch (IOException e) {
-                LOGGER.error("Failed to create SSLContext", e);
+                LOGGER.error("Failed to create SslContext", e);
            } catch (CertificateException e) {
-                LOGGER.error("Failed to create SSLContext", e);
-                throw new RuntimeException("Failed to create SSLContext", e);
+                LOGGER.error("Failed to create SslContext", e);
+                throw new RuntimeException("Failed to create SslContext", e);
            }
        }
    }
@@ -209,7 +209,7 @@ public class NettyRemotingClient extends NettyRemotingAbstract implements Remoti
                            pipeline.addFirst(defaultEventExecutorGroup, "sslHandler", sslContext.newHandler(ch.alloc()));
                            LOGGER.info("Prepend SSL handler");
                        } else {
-                            LOGGER.warn("Connections are insecure as SSLContext is null!");
+                            LOGGER.warn("Connections are insecure as SslContext is null!");
                        }
                    }
                    ch.pipeline().addLast(
@@ -337,7 +337,7 @@ public class NettyRemotingClient extends NettyRemotingAbstract implements Remoti
                                "sslHandler", sslContext.newHandler(ch.alloc()));
                            LOGGER.info("Prepend SSL handler");
                        } else {
-                            LOGGER.warn("Connections are insecure as SSLContext is null!");
+                            LOGGER.warn("Connections are insecure as SslContext is null!");
                        }
                    }

@@ -184,9 +184,9 @@ public class NettyRemotingServer extends NettyRemotingAbstract implements Remoti
        if (tlsMode != TlsMode.DISABLED) {
            try {
                sslContext = TlsHelper.buildSslContext(false);
-                log.info("SSLContext created for server");
+                log.info("SslContext created for server");
            } catch (CertificateException | IOException e) {
-                log.error("Failed to create SSLContext for server", e);
+                log.error("Failed to create SslContext for server", e);
            }
        }
    }
@@ -514,7 +514,7 @@ public class NettyRemotingServer extends NettyRemotingAbstract implements Remoti
                            log.info("Handlers prepended to channel pipeline to establish SSL connection");
                        } else {
                            ctx.close();
-                            log.error("Trying to establish an SSL connection but sslContext is null");
+                            log.error("Trying to establish an SSL connection but SslContext is null");
                        }
                        break;

@@ -34,15 +34,20 @@ import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;

 public class FileWatchService extends LifecycleAwareServiceThread {
    private static final Logger log = LoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME);
+    private static final int DEFAULT_WATCH_INTERVAL = 500;

    private final Map<String, String> currentHash = new HashMap<>();
    private final Listener listener;
-    private static final int WATCH_INTERVAL = 500;
+    private final int watchInterval;
    private final MessageDigest md = MessageDigest.getInstance("MD5");

-    public FileWatchService(final String[] watchFiles,
-        final Listener listener) throws Exception {
+    public FileWatchService(final String[] watchFiles, final Listener listener) throws Exception {
+        this(watchFiles, listener, DEFAULT_WATCH_INTERVAL);
+    }
+
+    public FileWatchService(final String[] watchFiles, final Listener listener, int watchInterval) throws Exception {
        this.listener = listener;
+        this.watchInterval = watchInterval;
        for (String file : watchFiles) {
            if (!Strings.isNullOrEmpty(file) && new File(file).exists()) {
                currentHash.put(file, md5Digest(file));
@@ -61,7 +66,7 @@ public class FileWatchService extends LifecycleAwareServiceThread {

        while (!this.isStopped()) {
            try {
-                this.waitForRunning(WATCH_INTERVAL);
+                this.waitForRunning(watchInterval);
                for (Map.Entry<String, String> entry : currentHash.entrySet()) {
                    String newHash = md5Digest(entry.getKey());
                    if (!newHash.equals(entry.getValue())) {