From 8fde5d3c8e9f6f9360a291eafef7869ff7dccb76 Mon Sep 17 00:00:00 2001 From: Jiwei Guo Date: Tue, 25 Aug 2026 10:27:43 +0800 Subject: [PATCH 1/4] git checkout -b fix-m1 --- .../channel/MQTTChannelInitializer.java | 53 ++---------------- mqtt-common/pom.xml | 14 ++++- mqtt-proxy/pom.xml | 4 -- .../channel/MQTTProxyChannelInitializer.java | 54 ++----------------- .../handlers/mqtt/proxy/web/WebService.java | 7 --- .../mtls/AuthenticationProviderMTlsTest.java | 4 +- pom.xml | 45 ++++++++++++++++ .../mqtt/mqtt5/hivemq/base/ProxyMtlsTest.java | 7 +-- 8 files changed, 73 insertions(+), 115 deletions(-) diff --git a/mqtt-broker/src/main/java/io/streamnative/pulsar/handlers/mqtt/broker/channel/MQTTChannelInitializer.java b/mqtt-broker/src/main/java/io/streamnative/pulsar/handlers/mqtt/broker/channel/MQTTChannelInitializer.java index d2a37a7fe..13db0a9ed 100644 --- a/mqtt-broker/src/main/java/io/streamnative/pulsar/handlers/mqtt/broker/channel/MQTTChannelInitializer.java +++ b/mqtt-broker/src/main/java/io/streamnative/pulsar/handlers/mqtt/broker/channel/MQTTChannelInitializer.java @@ -25,17 +25,14 @@ import io.streamnative.pulsar.handlers.mqtt.common.adapter.MqttAdapterDecoder; import io.streamnative.pulsar.handlers.mqtt.common.adapter.MqttAdapterEncoder; import io.streamnative.pulsar.handlers.mqtt.common.psk.PSKUtils; +import io.streamnative.pulsar.handlers.mqtt.common.tls.MQTTTlsFactory; import io.streamnative.pulsar.handlers.mqtt.common.utils.WebSocketUtils; import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; -import lombok.extern.slf4j.Slf4j; -import org.apache.pulsar.common.util.PulsarSslConfiguration; -import org.apache.pulsar.common.util.PulsarSslFactory; +import org.apache.pulsar.tls.TlsPurpose; /** * A channel initializer that initialize channels for MQTT protocol. */ -@Slf4j public class MQTTChannelInitializer extends ChannelInitializer { private final MQTTServerConfiguration mqttConfig; @@ -43,7 +40,7 @@ public class MQTTChannelInitializer extends ChannelInitializer { private final boolean enableTls; private final boolean enableTlsPsk; private final boolean enableWs; - private PulsarSslFactory sslFactory; + private MQTTTlsFactory tlsFactory; public MQTTChannelInitializer(MQTTService mqttService, boolean enableTls, boolean enableWs, ScheduledExecutorService sslContextRefresher) throws Exception { @@ -60,16 +57,7 @@ public MQTTChannelInitializer( this.enableTlsPsk = enableTlsPsk; this.enableWs = enableWs; if (this.enableTls) { - PulsarSslConfiguration sslConfiguration = buildSslConfiguration(mqttConfig); - this.sslFactory = (PulsarSslFactory) Class.forName(mqttConfig.getSslFactoryPlugin()) - .getConstructor().newInstance(); - this.sslFactory.initialize(sslConfiguration); - this.sslFactory.createInternalSslContext(); - if (mqttConfig.getTlsCertRefreshCheckDurationSec() > 0) { - sslContextRefresher.scheduleWithFixedDelay(this::refreshSslContext, - mqttConfig.getTlsCertRefreshCheckDurationSec(), - mqttConfig.getTlsCertRefreshCheckDurationSec(), TimeUnit.SECONDS); - } + this.tlsFactory = new MQTTTlsFactory(mqttConfig, TlsPurpose.BROKER, sslContextRefresher); } } @@ -77,7 +65,7 @@ public MQTTChannelInitializer( public void initChannel(SocketChannel ch) throws Exception { ch.pipeline().addFirst("idleStateHandler", new IdleStateHandler(0, 0, 120)); if (this.enableTls) { - ch.pipeline().addLast(TLS_HANDLER, new SslHandler(sslFactory.createServerSslEngine(ch.alloc()))); + ch.pipeline().addLast(TLS_HANDLER, tlsFactory.newServerSslHandler(ch.alloc())); } else if (this.enableTlsPsk) { ch.pipeline().addLast(TLS_HANDLER, new SslHandler(PSKUtils.createServerEngine(ch, mqttService.getPskConfiguration()))); @@ -95,35 +83,4 @@ public void initChannel(SocketChannel ch) throws Exception { ch.pipeline().addLast(MQTTBrokerInboundHandler.NAME, new MQTTBrokerInboundHandler(mqttService)); } - protected PulsarSslConfiguration buildSslConfiguration(MQTTServerConfiguration config) { - return PulsarSslConfiguration.builder() - .tlsProvider(config.getMqttTlsProvider()) - .tlsKeyStoreType(config.getMqttTlsKeyStoreType()) - .tlsKeyStorePath(config.getMqttTlsKeyStore()) - .tlsKeyStorePassword(config.getMqttTlsKeyStorePassword()) - .tlsTrustStoreType(config.getMqttTlsTrustStoreType()) - .tlsTrustStorePath(config.getMqttTlsTrustStore()) - .tlsTrustStorePassword(config.getMqttTlsTrustStorePassword()) - .tlsCiphers(config.getMqttTlsCiphers()) - .tlsProtocols(config.getMqttTlsProtocols()) - .tlsTrustCertsFilePath(config.getMqttTlsTrustCertsFilePath()) - .tlsCertificateFilePath(config.getMqttTlsCertificateFilePath()) - .tlsKeyFilePath(config.getMqttTlsKeyFilePath()) - .allowInsecureConnection(config.isMqttTlsAllowInsecureConnection()) - .requireTrustedClientCertOnConnect(config.isMqttTlsRequireTrustedClientCertOnConnect()) - .tlsEnabledWithKeystore(config.isMqttTlsEnabledWithKeyStore()) - .tlsCustomParams(config.getSslFactoryPluginParams()) - .authData(null) - .serverMode(true) - .build(); - } - - protected void refreshSslContext() { - try { - this.sslFactory.update(); - } catch (Exception e) { - log.error("Failed to refresh SSL context for mqtt channel.", e); - } - } - } diff --git a/mqtt-common/pom.xml b/mqtt-common/pom.xml index 51a09e369..36af7f589 100644 --- a/mqtt-common/pom.xml +++ b/mqtt-common/pom.xml @@ -30,6 +30,18 @@ io.streamnative pulsar-client-original + + io.streamnative + pulsar-tls-factory-api + + + io.streamnative + pulsar-common + + + io.streamnative + pulsar-broker-common + org.apache.bookkeeper bookkeeper-common @@ -49,7 +61,6 @@ javax.validation validation-api - ${validation-api.version} com.github.ben-manes.caffeine @@ -58,7 +69,6 @@ org.conscrypt conscrypt-openjdk-uber - ${conscrypt.version} diff --git a/mqtt-proxy/pom.xml b/mqtt-proxy/pom.xml index 9e4ed8d69..0d116e4bc 100644 --- a/mqtt-proxy/pom.xml +++ b/mqtt-proxy/pom.xml @@ -43,7 +43,6 @@ org.eclipse.jetty.ee10 jetty-ee10-servlet - ${jetty.ee10.version} org.glassfish.jersey.core @@ -52,17 +51,14 @@ org.glassfish.jersey.containers jersey-container-servlet-core - ${jersey.version} org.glassfish.jersey.media jersey-media-multipart - ${jersey.version} io.swagger swagger-annotations - ${swagger-annotations.version} jakarta.ws.rs diff --git a/mqtt-proxy/src/main/java/io/streamnative/pulsar/handlers/mqtt/proxy/channel/MQTTProxyChannelInitializer.java b/mqtt-proxy/src/main/java/io/streamnative/pulsar/handlers/mqtt/proxy/channel/MQTTProxyChannelInitializer.java index 6bb14a395..19f74dcf5 100644 --- a/mqtt-proxy/src/main/java/io/streamnative/pulsar/handlers/mqtt/proxy/channel/MQTTProxyChannelInitializer.java +++ b/mqtt-proxy/src/main/java/io/streamnative/pulsar/handlers/mqtt/proxy/channel/MQTTProxyChannelInitializer.java @@ -23,21 +23,18 @@ import io.streamnative.pulsar.handlers.mqtt.common.adapter.MqttAdapterDecoder; import io.streamnative.pulsar.handlers.mqtt.common.adapter.MqttAdapterEncoder; import io.streamnative.pulsar.handlers.mqtt.common.psk.PSKUtils; +import io.streamnative.pulsar.handlers.mqtt.common.tls.MQTTTlsFactory; import io.streamnative.pulsar.handlers.mqtt.common.utils.WebSocketUtils; import io.streamnative.pulsar.handlers.mqtt.proxy.MQTTProxyConfiguration; import io.streamnative.pulsar.handlers.mqtt.proxy.MQTTProxyService; import io.streamnative.pulsar.handlers.mqtt.proxy.impl.MQTTProxyException; import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; import lombok.Getter; -import lombok.extern.slf4j.Slf4j; -import org.apache.pulsar.common.util.PulsarSslConfiguration; -import org.apache.pulsar.common.util.PulsarSslFactory; +import org.apache.pulsar.tls.TlsPurpose; /** * Proxy service channel initializer. */ -@Slf4j public class MQTTProxyChannelInitializer extends ChannelInitializer { private final MQTTProxyService proxyService; @@ -47,7 +44,7 @@ public class MQTTProxyChannelInitializer extends ChannelInitializer 0) { - sslContextRefresher.scheduleWithFixedDelay(this::refreshSslContext, - proxyConfig.getTlsCertRefreshCheckDurationSec(), - proxyConfig.getTlsCertRefreshCheckDurationSec(), TimeUnit.SECONDS); - - } + this.tlsFactory = new MQTTTlsFactory(proxyConfig, TlsPurpose.PROXY, sslContextRefresher); } } catch (Exception e) { throw new MQTTProxyException(e); @@ -86,7 +73,7 @@ public MQTTProxyChannelInitializer(MQTTProxyService proxyService, MQTTProxyConfi protected void initChannel(SocketChannel ch) throws Exception { ch.pipeline().addFirst("idleStateHandler", new IdleStateHandler(30, 0, 0)); if (this.enableTls) { - ch.pipeline().addLast(TLS_HANDLER, new SslHandler(sslFactory.createServerSslEngine(ch.alloc()))); + ch.pipeline().addLast(TLS_HANDLER, tlsFactory.newServerSslHandler(ch.alloc())); } else if (this.enableTlsPsk) { ch.pipeline().addLast(TLS_HANDLER, new SslHandler(PSKUtils.createServerEngine(ch, proxyService.getPskConfiguration()))); @@ -103,35 +90,4 @@ protected void initChannel(SocketChannel ch) throws Exception { ch.pipeline().addLast("handler", new MQTTProxyInboundHandler(proxyService)); } - protected PulsarSslConfiguration buildSslConfiguration(MQTTProxyConfiguration config) { - return PulsarSslConfiguration.builder() - .tlsProvider(config.getMqttTlsProvider()) - .tlsKeyStoreType(config.getMqttTlsKeyStoreType()) - .tlsKeyStorePath(config.getMqttTlsKeyStore()) - .tlsKeyStorePassword(config.getMqttTlsKeyStorePassword()) - .tlsTrustStoreType(config.getMqttTlsTrustStoreType()) - .tlsTrustStorePath(config.getMqttTlsTrustStore()) - .tlsTrustStorePassword(config.getMqttTlsTrustStorePassword()) - .tlsCiphers(config.getMqttTlsCiphers()) - .tlsProtocols(config.getMqttTlsProtocols()) - .tlsTrustCertsFilePath(config.getMqttTlsTrustCertsFilePath()) - .tlsCertificateFilePath(config.getMqttTlsCertificateFilePath()) - .tlsKeyFilePath(config.getMqttTlsKeyFilePath()) - .allowInsecureConnection(config.isMqttTlsAllowInsecureConnection()) - .requireTrustedClientCertOnConnect(config.isMqttTlsRequireTrustedClientCertOnConnect()) - .tlsEnabledWithKeystore(config.isMqttTlsEnabledWithKeyStore()) - .tlsCustomParams(config.getSslFactoryPluginParams()) - .authData(null) - .serverMode(true) - .build(); - } - - protected void refreshSslContext() { - try { - this.sslFactory.update(); - } catch (Exception e) { - log.error("Failed to refresh SSL context for mqtt proxy channel.", e); - } - } - } diff --git a/mqtt-proxy/src/main/java/io/streamnative/pulsar/handlers/mqtt/proxy/web/WebService.java b/mqtt-proxy/src/main/java/io/streamnative/pulsar/handlers/mqtt/proxy/web/WebService.java index 731d14c7e..98a973d15 100644 --- a/mqtt-proxy/src/main/java/io/streamnative/pulsar/handlers/mqtt/proxy/web/WebService.java +++ b/mqtt-proxy/src/main/java/io/streamnative/pulsar/handlers/mqtt/proxy/web/WebService.java @@ -22,7 +22,6 @@ import java.util.List; import java.util.Map; import java.util.Optional; -import java.util.concurrent.ScheduledFuture; import lombok.Getter; import org.apache.pulsar.broker.web.DynamicSkipUnknownPropertyHandler; import org.apache.pulsar.broker.web.GzipHandlerUtil; @@ -30,7 +29,6 @@ import org.apache.pulsar.broker.web.JsonMapperProvider; import org.apache.pulsar.broker.web.UnrecognizedPropertyExceptionMapper; import org.apache.pulsar.broker.web.WebExecutorThreadPool; -import org.apache.pulsar.common.util.PulsarSslFactory; import org.apache.pulsar.jetty.metrics.JettyStatisticsCollector; import org.eclipse.jetty.ee10.servlet.ServletContextHandler; import org.eclipse.jetty.ee10.servlet.ServletHolder; @@ -73,8 +71,6 @@ public class WebService implements AutoCloseable { private final ServerConnector httpConnector; private JettyStatisticsCollector jettyStatisticsCollector; - private PulsarSslFactory sslFactory; - private ScheduledFuture sslContextRefreshTask; private final MQTTCommonConfiguration config; private final MQTTProxyService proxyService; @@ -259,9 +255,6 @@ public void close() throws MQTTProxyException { jettyStatisticsCollector = null; } webServiceExecutor.join(); - if (this.sslContextRefreshTask != null) { - this.sslContextRefreshTask.cancel(true); - } log.info("Web service closed"); } catch (Exception e) { throw new MQTTProxyException(e); diff --git a/mqtt-proxy/src/test/java/io/streamnative/pulsar/handlers/mqtt/proxy/authentication/mtls/AuthenticationProviderMTlsTest.java b/mqtt-proxy/src/test/java/io/streamnative/pulsar/handlers/mqtt/proxy/authentication/mtls/AuthenticationProviderMTlsTest.java index 7102847d0..b3c37876b 100644 --- a/mqtt-proxy/src/test/java/io/streamnative/pulsar/handlers/mqtt/proxy/authentication/mtls/AuthenticationProviderMTlsTest.java +++ b/mqtt-proxy/src/test/java/io/streamnative/pulsar/handlers/mqtt/proxy/authentication/mtls/AuthenticationProviderMTlsTest.java @@ -31,7 +31,7 @@ import org.apache.pulsar.broker.ServiceConfiguration; import org.apache.pulsar.broker.authentication.AuthenticationDataCommand; import org.apache.pulsar.common.util.ObjectMapperFactory; -import org.apache.pulsar.common.util.SecurityUtility; +import org.apache.pulsar.common.util.tls.PemReader; import org.apache.pulsar.metadata.api.MetadataStoreConfig; import org.apache.pulsar.metadata.impl.LocalMemoryMetadataStore; import org.awaitility.Awaitility; @@ -122,7 +122,7 @@ public void testAuthenticationProviderMTls(boolean reuseMetadata) throws Excepti Awaitility.await().until(() -> authenticationProvider.getPoolMap().size() == 1); X509Certificate[] x509Certificates = - SecurityUtility.loadCertificatesFromPemFile(getResourcePath("mtls/client-cert.pem")); + PemReader.loadCertificatesFromPemFile(getResourcePath("mtls/client-cert.pem")); SSLSession sslSession = new MockSSLSession(x509Certificates); AuthenticationDataCommand authData = new AuthenticationDataCommand("", LocalAddress.ANY, sslSession); diff --git a/pom.xml b/pom.xml index 24d27c979..58bce4520 100644 --- a/pom.xml +++ b/pom.xml @@ -160,6 +160,51 @@ pom import + + io.streamnative + pulsar-tls-factory-api + ${pulsar.version} + + + io.streamnative + pulsar-common + ${pulsar.version} + + + io.streamnative + pulsar-broker-common + ${pulsar.version} + + + javax.validation + validation-api + ${validation-api.version} + + + org.conscrypt + conscrypt-openjdk-uber + ${conscrypt.version} + + + org.eclipse.jetty.ee10 + jetty-ee10-servlet + ${jetty.ee10.version} + + + org.glassfish.jersey.containers + jersey-container-servlet-core + ${jersey.version} + + + org.glassfish.jersey.media + jersey-media-multipart + ${jersey.version} + + + io.swagger + swagger-annotations + ${swagger-annotations.version} + diff --git a/tests/src/test/java/io/streamnative/pulsar/handlers/mqtt/mqtt5/hivemq/base/ProxyMtlsTest.java b/tests/src/test/java/io/streamnative/pulsar/handlers/mqtt/mqtt5/hivemq/base/ProxyMtlsTest.java index 3c4a12acd..150b8c9bc 100644 --- a/tests/src/test/java/io/streamnative/pulsar/handlers/mqtt/mqtt5/hivemq/base/ProxyMtlsTest.java +++ b/tests/src/test/java/io/streamnative/pulsar/handlers/mqtt/mqtt5/hivemq/base/ProxyMtlsTest.java @@ -50,7 +50,8 @@ import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.impl.auth.AuthenticationToken; -import org.apache.pulsar.common.util.SecurityUtility; +import org.apache.pulsar.common.util.tls.JdkSslContexts; +import org.apache.pulsar.common.util.tls.PemReader; import org.awaitility.Awaitility; import org.fusesource.mqtt.client.BlockingConnection; import org.fusesource.mqtt.client.MQTT; @@ -146,9 +147,9 @@ public SSLContext createSSLContext() throws Exception { Certificate clientCert = CertificateFactory .getInstance("X.509").generateCertificate(new FileInputStream(clientCertFile)); - PrivateKey privateKey = SecurityUtility.loadPrivateKeyFromPemFile(path + "client.key"); + PrivateKey privateKey = PemReader.loadPrivateKeyFromPemFile(path + "client.key"); - final SSLContext sslContext = SecurityUtility.createSslContext(true, + final SSLContext sslContext = JdkSslContexts.createSslContext(true, new Certificate[]{caCert}, new Certificate[]{clientCert}, privateKey); return sslContext; From 0dab66c8bbd31580c1f7576a4b89973f20f0ff21 Mon Sep 17 00:00:00 2001 From: Jiwei Guo Date: Tue, 25 Aug 2026 10:29:04 +0800 Subject: [PATCH 2/4] git checkout -b fix-m1 --- .../mqtt/common/tls/MQTTTlsFactory.java | 124 ++++++++++++++++++ 1 file changed, 124 insertions(+) create mode 100644 mqtt-common/src/main/java/io/streamnative/pulsar/handlers/mqtt/common/tls/MQTTTlsFactory.java diff --git a/mqtt-common/src/main/java/io/streamnative/pulsar/handlers/mqtt/common/tls/MQTTTlsFactory.java b/mqtt-common/src/main/java/io/streamnative/pulsar/handlers/mqtt/common/tls/MQTTTlsFactory.java new file mode 100644 index 000000000..c08e984ac --- /dev/null +++ b/mqtt-common/src/main/java/io/streamnative/pulsar/handlers/mqtt/common/tls/MQTTTlsFactory.java @@ -0,0 +1,124 @@ +/** + * Licensed 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 io.streamnative.pulsar.handlers.mqtt.common.tls; + +import io.netty.buffer.ByteBufAllocator; +import io.netty.handler.ssl.SslContext; +import io.netty.handler.ssl.SslHandler; +import io.streamnative.pulsar.handlers.mqtt.common.MQTTCommonConfiguration; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ScheduledExecutorService; +import org.apache.pulsar.broker.tls.TlsFactorySupport; +import org.apache.pulsar.common.tls.impl.FileBasedTlsFactory; +import org.apache.pulsar.common.tls.impl.FileBasedTlsFactorySettings; +import org.apache.pulsar.common.tls.impl.TlsContextAcquisition; +import org.apache.pulsar.common.tls.impl.TlsSynthesisSpec; +import org.apache.pulsar.tls.PulsarTlsFactory; +import org.apache.pulsar.tls.TlsFactoryInitContext; +import org.apache.pulsar.tls.TlsHandle; +import org.apache.pulsar.tls.TlsPolicy; +import org.apache.pulsar.tls.TlsPurpose; + +/** + * Bridges MoP's mqttTls* settings to Pulsar's PIP-478 TLS factory SPI. + */ +public class MQTTTlsFactory implements AutoCloseable { + + private PulsarTlsFactory tlsFactory; + private TlsHandle tlsSubscription; + private volatile SslContext tlsServerContext; + + public MQTTTlsFactory(MQTTCommonConfiguration config, TlsPurpose purpose, + ScheduledExecutorService sslContextRefresher) throws Exception { + this.tlsFactory = TlsFactorySupport.createFactory(config.getTlsFactoryClassName(), + FileBasedTlsFactory.class, + () -> createDefaultFactory(config, purpose)); + try { + TlsFactoryInitContext initContext = TlsFactorySupport.initContext( + TlsFactorySupport.parseFactoryConfig(config.getTlsFactoryConfig()), + sslContextRefresher, sslContextRefresher); + TlsFactorySupport.initializeBlocking(this.tlsFactory, initContext); + this.tlsSubscription = TlsContextAcquisition.acquireNettyContext(this.tlsFactory, purpose, + TlsSynthesisSpec.server(config.isMqttTlsRequireTrustedClientCertOnConnect()), + context -> this.tlsServerContext = context) + .get() + .orElseThrow(() -> new IllegalStateException( + "TLS factory supplied no Netty SslContext for purpose " + purpose)); + } catch (Exception e) { + close(); + throw e; + } + } + + public SslHandler newServerSslHandler(ByteBufAllocator allocator) { + return TlsContextAcquisition.withPinnedContext( + () -> this.tlsServerContext, context -> context.newHandler(allocator)); + } + + @Override + public void close() { + TlsHandle subscription = this.tlsSubscription; + if (subscription != null) { + this.tlsSubscription = null; + subscription.dispose(); + } + PulsarTlsFactory factory = this.tlsFactory; + if (factory != null) { + this.tlsFactory = null; + factory.close(); + } + } + + private static PulsarTlsFactory createDefaultFactory(MQTTCommonConfiguration config, TlsPurpose purpose) { + Map policies = Map.of(purpose, createPolicy(config)); + FileBasedTlsFactorySettings settings = FileBasedTlsFactorySettings.builder() + .requireTrustedClientCert(config.isMqttTlsRequireTrustedClientCertOnConnect()) + .refreshIntervalSeconds(FileBasedTlsFactorySettings.refreshIntervalSecondsFromConfig( + config.getMqttTlsCertRefreshCheckDurationSec())) + .engineProvider(TlsFactorySupport.engineProvider(config.getMqttTlsProvider())) + .build(); + return new FileBasedTlsFactory(policies, settings); + } + + private static TlsPolicy createPolicy(MQTTCommonConfiguration config) { + TlsPolicy.Builder builder = TlsPolicy.builder() + .allowInsecureConnection(config.isMqttTlsAllowInsecureConnection()) + .enableHostnameVerification(config.isTlsHostnameVerificationEnabled()) + .protocols(toList(config.getMqttTlsProtocols())) + .ciphers(toList(config.getMqttTlsCiphers())) + .jsseProvider(TlsFactorySupport.resolveJsseProvider(config.getJsseProvider(), + config.getMqttTlsProvider())); + if (config.isMqttTlsEnabledWithKeyStore()) { + builder.format(TlsPolicy.Format.KEYSTORE) + .keyStoreType(config.getMqttTlsKeyStoreType()) + .trustStoreType(config.getMqttTlsTrustStoreType()) + .keyStorePath(config.getMqttTlsKeyStore()) + .keyStorePassword(config.getMqttTlsKeyStorePassword()) + .trustStorePath(config.getMqttTlsTrustStore()) + .trustStorePassword(config.getMqttTlsTrustStorePassword()); + } else { + builder.format(TlsPolicy.Format.PEM) + .trustCertsFilePath(config.getMqttTlsTrustCertsFilePath()) + .certificateFilePath(config.getMqttTlsCertificateFilePath()) + .keyFilePath(config.getMqttTlsKeyFilePath()); + } + return builder.build(); + } + + private static List toList(Set values) { + return values == null ? List.of() : List.copyOf(values); + } +} From 0e77b8941f31de45b8b40132c0465f4960eda490 Mon Sep 17 00:00:00 2001 From: Jiwei Guo Date: Tue, 25 Aug 2026 10:34:23 +0800 Subject: [PATCH 3/4] fix checkstyle --- .../handlers/mqtt/common/tls/package-info.java | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) create mode 100644 mqtt-common/src/main/java/io/streamnative/pulsar/handlers/mqtt/common/tls/package-info.java diff --git a/mqtt-common/src/main/java/io/streamnative/pulsar/handlers/mqtt/common/tls/package-info.java b/mqtt-common/src/main/java/io/streamnative/pulsar/handlers/mqtt/common/tls/package-info.java new file mode 100644 index 000000000..177ade5f5 --- /dev/null +++ b/mqtt-common/src/main/java/io/streamnative/pulsar/handlers/mqtt/common/tls/package-info.java @@ -0,0 +1,18 @@ +/** + * Licensed 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. + */ + +/** + * TLS support for MQTT protocol handlers. + */ +package io.streamnative.pulsar.handlers.mqtt.common.tls; From 53f1aec4d2bc316fe8105c37e2229bb7ffaa7a3a Mon Sep 17 00:00:00 2001 From: Jiwei Guo Date: Tue, 25 Aug 2026 10:45:35 +0800 Subject: [PATCH 4/4] fix --- pom.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pom.xml b/pom.xml index 58bce4520..707e645c5 100644 --- a/pom.xml +++ b/pom.xml @@ -71,7 +71,7 @@ 3.1.8 0.8.7 2.0.1.Final - 3.1.10 + 3.1.12 12.1.10 1.6.15 2.5.2