This is an automated email from the ASF dual-hosted git repository. lhotari pushed a commit to branch branch-3.1 in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 628e79ded04b8aeb0e146e717e8ed5442c566a6b Author: Jiwei Guo <[email protected]> AuthorDate: Sun Sep 17 20:44:13 2023 -0500 [fix][broker] Fix PulsarService.getLookupServiceAddress returns wrong port if TLS is enabled (#21015) (cherry picked from commit 1363777918547b0f56bdbbf04cec28d050a43586) --- .../org/apache/pulsar/broker/PulsarService.java | 10 +++--- .../apache/pulsar/broker/PulsarServiceTest.java | 2 ++ .../apache/pulsar/broker/admin/AdminApi2Test.java | 14 +++----- .../apache/pulsar/broker/admin/AdminApiTest.java | 2 +- .../broker/auth/MockedPulsarServiceBaseTest.java | 4 --- .../loadbalance/AdvertisedListenersTest.java | 2 -- .../broker/loadbalance/LoadBalancerTest.java | 1 - .../loadbalance/SimpleLoadManagerImplTest.java | 1 - .../impl/ModularLoadManagerImplTest.java | 6 ---- .../OwnerShipForCurrentServerTestBase.java | 2 -- .../pulsar/broker/service/BrokerServiceTest.java | 9 +++++ .../broker/service/ClusterMigrationTest.java | 9 +++++ .../pulsar/broker/service/ReplicatorTest.java | 2 +- .../broker/transaction/TransactionTestBase.java | 2 -- .../pulsar/client/api/BrokerServiceLookupTest.java | 7 +++- .../client/api/ClientAuthenticationTlsTest.java | 4 ++- .../pulsar/proxy/server/ProxyRefreshAuthTest.java | 1 + .../proxy/server/ProxyServiceTlsStarterTest.java | 1 + .../integration/containers/BrokerContainer.java | 9 +++-- .../integration/containers/ProxyContainer.java | 9 +++-- .../tests/integration/tls/ClientTlsTest.java | 9 +++++ .../integration/topologies/PulsarCluster.java | 40 +++++++++++++++------- .../integration/topologies/PulsarClusterSpec.java | 6 ++++ 23 files changed, 98 insertions(+), 54 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java index a366cf25aa0..50696bda37c 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java @@ -1733,18 +1733,18 @@ public class PulsarService implements AutoCloseable, ShutdownService { } public String getSafeWebServiceAddress() { - return webServiceAddress != null ? webServiceAddress : webServiceAddressTls; + return webServiceAddressTls != null ? webServiceAddressTls : webServiceAddress; } @Deprecated public String getSafeBrokerServiceUrl() { - return brokerServiceUrl != null ? brokerServiceUrl : brokerServiceUrlTls; + return brokerServiceUrlTls != null ? brokerServiceUrlTls : brokerServiceUrl; } public String getLookupServiceAddress() { - return String.format("%s:%s", advertisedAddress, config.getWebServicePort().isPresent() - ? config.getWebServicePort().get() - : config.getWebServicePortTls().orElseThrow()); + return String.format("%s:%s", advertisedAddress, config.getWebServicePortTls().isPresent() + ? config.getWebServicePortTls().get() + : config.getWebServicePort().orElseThrow()); } public TopicPoliciesService getTopicPoliciesService() { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/PulsarServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/PulsarServiceTest.java index 37a7310ae17..3e0887646e1 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/PulsarServiceTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/PulsarServiceTest.java @@ -54,6 +54,8 @@ public class PulsarServiceTest extends MockedPulsarServiceBaseTest { @Override protected void doInitConf() throws Exception { super.doInitConf(); + conf.setBrokerServicePortTls(Optional.of(0)); + conf.setWebServicePortTls(Optional.of(0)); if (useStaticPorts) { conf.setBrokerServicePortTls(Optional.of(6651)); conf.setBrokerServicePort(Optional.of(6660)); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java index 00134f8c918..723d576dea3 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java @@ -439,19 +439,13 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest { String tenantName = newUniqueName("prop-xyz2"); admin.tenants().createTenant(tenantName, tenantInfo); admin.namespaces().createNamespace(tenantName + "/ns1", Set.of("test")); - conf.setBrokerServicePort(Optional.of(1024)); - conf.setBrokerServicePortTls(Optional.of(1025)); - conf.setWebServicePort(Optional.of(1026)); - conf.setWebServicePortTls(Optional.of(1027)); + ServiceConfiguration config2 = super.getDefaultConf(); @Cleanup - PulsarTestContext pulsarTestContext2 = createAdditionalPulsarTestContext(conf); + PulsarTestContext pulsarTestContext2 = createAdditionalPulsarTestContext(config2); PulsarService pulsar2 = pulsarTestContext2.getPulsarService(); - conf.setBrokerServicePort(Optional.of(2048)); - conf.setBrokerServicePortTls(Optional.of(2049)); - conf.setWebServicePort(Optional.of(2050)); - conf.setWebServicePortTls(Optional.of(2051)); + ServiceConfiguration config3 = super.getDefaultConf(); @Cleanup - PulsarTestContext pulsarTestContext3 = createAdditionalPulsarTestContext(conf); + PulsarTestContext pulsarTestContext3 = createAdditionalPulsarTestContext(config3); PulsarService pulsar3 = pulsarTestContext.getPulsarService(); @Cleanup PulsarAdmin admin2 = PulsarAdmin.builder().serviceHttpUrl(pulsar2.getWebServiceAddress()).build(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java index 9fb3cb6fb23..e97707710d7 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java @@ -931,7 +931,7 @@ public class AdminApiTest extends MockedPulsarServiceBaseTest { assertEquals(topicStats.getSubscriptions().get(subName).getMsgBacklog(), 10); assertEquals(topicStats.getPublishers().size(), 0); assertEquals(topicStats.getOwnerBroker(), - pulsar.getAdvertisedAddress() + ":" + pulsar.getConfiguration().getWebServicePort().get()); + pulsar.getAdvertisedAddress() + ":" + pulsar.getConfiguration().getWebServicePortTls().get()); PersistentTopicInternalStats internalStats = admin.topics().getInternalStats(persistentTopicName, false); assertEquals(internalStats.cursors.keySet(), Set.of(Codec.encode(subName))); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java index 941a9356a86..cb51b8aee35 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java @@ -233,10 +233,8 @@ public abstract class MockedPulsarServiceBaseTest extends TestRetrySupport { this.conf.setBrokerShutdownTimeoutMs(0L); this.conf.setLoadBalancerOverrideBrokerNicSpeedGbps(Optional.of(1.0d)); this.conf.setBrokerServicePort(Optional.of(0)); - this.conf.setBrokerServicePortTls(Optional.of(0)); this.conf.setAdvertisedAddress("localhost"); this.conf.setWebServicePort(Optional.of(0)); - this.conf.setWebServicePortTls(Optional.of(0)); this.conf.setNumExecutorThreadPoolSize(5); this.conf.setExposeBundlesMetricsInPrometheus(true); } @@ -492,9 +490,7 @@ public abstract class MockedPulsarServiceBaseTest extends TestRetrySupport { configuration.setBrokerShutdownTimeoutMs(0L); configuration.setLoadBalancerOverrideBrokerNicSpeedGbps(Optional.of(1.0d)); configuration.setBrokerServicePort(Optional.of(0)); - configuration.setBrokerServicePortTls(Optional.of(0)); configuration.setWebServicePort(Optional.of(0)); - configuration.setWebServicePortTls(Optional.of(0)); configuration.setBookkeeperClientExposeStatsToPrometheus(true); configuration.setNumExecutorThreadPoolSize(5); configuration.setBrokerMaxConnections(0); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/AdvertisedListenersTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/AdvertisedListenersTest.java index 7a8154312e4..a88ccd60ae4 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/AdvertisedListenersTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/AdvertisedListenersTest.java @@ -78,7 +78,6 @@ public class AdvertisedListenersTest extends MultiBrokerBaseTest { ",public_https:https://localhost:"; + httpsPort); conf.setBrokerServicePort(Optional.of(pulsarPort)); conf.setWebServicePort(Optional.of(httpPort)); - conf.setWebServicePortTls(Optional.of(httpsPort)); } @Test @@ -101,7 +100,6 @@ public class AdvertisedListenersTest extends MultiBrokerBaseTest { assertEquals(new URI(ld.getBrokerUrl()).getHost(), "localhost"); assertEquals(new URI(ld.getHttpUrl()).getHost(), "localhost"); - assertEquals(new URI(ld.getHttpUrlTls()).getHost(), "localhost"); // Produce data diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LoadBalancerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LoadBalancerTest.java index 68902c73e57..7cc4499df97 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LoadBalancerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LoadBalancerTest.java @@ -124,7 +124,6 @@ public class LoadBalancerTest { config.setAdvertisedAddress("localhost"); config.setWebServicePort(Optional.of(0)); config.setBrokerServicePortTls(Optional.of(0)); - config.setWebServicePortTls(Optional.of(0)); config.setMetadataStoreUrl("zk:127.0.0.1:" + bkEnsemble.getZookeeperPort()); config.setBrokerShutdownTimeoutMs(0L); config.setLoadBalancerOverrideBrokerNicSpeedGbps(Optional.of(1.0d)); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/SimpleLoadManagerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/SimpleLoadManagerImplTest.java index c4898786e3e..6303c70b4dc 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/SimpleLoadManagerImplTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/SimpleLoadManagerImplTest.java @@ -113,7 +113,6 @@ public class SimpleLoadManagerImplTest { config1.setBrokerServicePort(Optional.of(0)); config1.setLoadManagerClassName(SimpleLoadManagerImpl.class.getName()); config1.setBrokerServicePortTls(Optional.of(0)); - config1.setWebServicePortTls(Optional.of(0)); config1.setAdvertisedAddress("localhost"); pulsar1 = new PulsarService(config1); pulsar1.start(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java index e3206ba5167..ceeb5704fb2 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java @@ -172,8 +172,6 @@ public class ModularLoadManagerImplTest { config1.setBrokerShutdownTimeoutMs(0L); config1.setLoadBalancerOverrideBrokerNicSpeedGbps(Optional.of(1.0d)); config1.setBrokerServicePort(Optional.of(0)); - config1.setBrokerServicePortTls(Optional.of(0)); - config1.setWebServicePortTls(Optional.of(0)); pulsar1 = new PulsarService(config1); pulsar1.start(); @@ -192,8 +190,6 @@ public class ModularLoadManagerImplTest { config2.setBrokerShutdownTimeoutMs(0L); config2.setLoadBalancerOverrideBrokerNicSpeedGbps(Optional.of(1.0d)); config2.setBrokerServicePort(Optional.of(0)); - config2.setBrokerServicePortTls(Optional.of(0)); - config2.setWebServicePortTls(Optional.of(0)); pulsar2 = new PulsarService(config2); pulsar2.start(); @@ -207,8 +203,6 @@ public class ModularLoadManagerImplTest { config.setBrokerShutdownTimeoutMs(0L); config.setLoadBalancerOverrideBrokerNicSpeedGbps(Optional.of(1.0d)); config.setBrokerServicePort(Optional.of(0)); - config.setBrokerServicePortTls(Optional.of(0)); - config.setWebServicePortTls(Optional.of(0)); pulsar3 = new PulsarService(config); secondaryHost = String.format("%s:%d", "localhost", pulsar2.getListenPortHTTP().get()); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnerShipForCurrentServerTestBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnerShipForCurrentServerTestBase.java index 8dd4f53db82..46e8989ac3d 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnerShipForCurrentServerTestBase.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnerShipForCurrentServerTestBase.java @@ -80,10 +80,8 @@ public abstract class OwnerShipForCurrentServerTestBase { conf.setBrokerShutdownTimeoutMs(0L); conf.setLoadBalancerOverrideBrokerNicSpeedGbps(Optional.of(1.0d)); conf.setBrokerServicePort(Optional.of(0)); - conf.setBrokerServicePortTls(Optional.of(0)); conf.setAdvertisedAddress("localhost"); conf.setWebServicePort(Optional.of(0)); - conf.setWebServicePortTls(Optional.of(0)); serviceConfigurationList.add(conf); PulsarTestContext.Builder testContextBuilder = diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java index e0f2fadc2b6..43ea15a0391 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java @@ -1786,4 +1786,13 @@ public class BrokerServiceTest extends BrokerTestBase { fail("Unsubscribe failed"); } } + + @Test + public void testGetLookupServiceAddress() throws Exception { + cleanup(); + setup(); + conf.setWebServicePortTls(Optional.of(8081)); + assertEquals(pulsar.getLookupServiceAddress(), "localhost:8081"); + resetState(); + } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ClusterMigrationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ClusterMigrationTest.java index c4f2cd047af..a2dcf3c9c0b 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ClusterMigrationTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ClusterMigrationTest.java @@ -28,6 +28,7 @@ import static org.testng.Assert.assertTrue; import com.google.common.collect.Sets; import java.lang.reflect.Method; import java.net.URL; +import java.util.Optional; import java.util.concurrent.TimeUnit; import lombok.Cleanup; import org.apache.pulsar.broker.BrokerTestUtil; @@ -477,6 +478,14 @@ public class ClusterMigrationTest { super.setupWithClusterName(clusterName); } + @Override + protected void doInitConf() throws Exception { + super.doInitConf(); + this.conf.setWebServicePortTls(Optional.of(0)); + this.conf.setBrokerServicePortTls(Optional.of(0)); + } + + public PulsarService getPulsarService() { return pulsar; } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java index 88a6f7c9f69..c372e3029ab 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java @@ -248,7 +248,7 @@ public class ReplicatorTest extends ReplicatorTestBase { -> admin2.clusters().getCluster(cluster2) != null); List<String> list = admin1.brokers().getActiveBrokers(cluster2); - assertEquals(list.get(0), url2.toString().replace("http://";, "")); + assertEquals(list.get(0), urlTls2.toString().replace("https://";, "")); //restore configuration pulsar1.getConfiguration().setAuthorizationEnabled(false); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTestBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTestBase.java index c0300c63b35..1ff835732aa 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTestBase.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTestBase.java @@ -157,10 +157,8 @@ public abstract class TransactionTestBase extends TestRetrySupport { conf.setBrokerShutdownTimeoutMs(0L); conf.setLoadBalancerOverrideBrokerNicSpeedGbps(Optional.of(1.0d)); conf.setBrokerServicePort(Optional.of(0)); - conf.setBrokerServicePortTls(Optional.of(0)); conf.setAdvertisedAddress("localhost"); conf.setWebServicePort(Optional.of(0)); - conf.setWebServicePortTls(Optional.of(0)); conf.setTransactionCoordinatorEnabled(true); conf.setBrokerDeduplicationEnabled(true); conf.setTransactionBufferSnapshotMaxTransactionCount(2); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java index c0cc06795ac..6becc9cb578 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java @@ -479,7 +479,7 @@ public class BrokerServiceLookupTest extends ProducerConsumerBase { // request [3] doReturn(true).when(loadManager1).isCentralized(); doReturn(true).when(loadManager2).isCentralized(); - SimpleResourceUnit resourceUnit = new SimpleResourceUnit(pulsar.getWebServiceAddress(), null); + SimpleResourceUnit resourceUnit = new SimpleResourceUnit(pulsar.getWebServiceAddressTls(), null); doReturn(Optional.of(resourceUnit)).when(loadManager2).getLeastLoaded(any(ServiceUnitId.class)); doReturn(Optional.of(resourceUnit)).when(loadManager1).getLeastLoaded(any(ServiceUnitId.class)); @@ -512,6 +512,9 @@ public class BrokerServiceLookupTest extends ProducerConsumerBase { loadManager1 = null; loadManager2 = null; + + conf.setBrokerServicePortTls(Optional.empty()); + conf.setWebServicePortTls(Optional.empty()); } /** @@ -943,6 +946,8 @@ public class BrokerServiceLookupTest extends ProducerConsumerBase { admin.topics().createPartitionedTopic(dest.toString(), totalPartitions); stopBroker(); + conf.setBrokerServicePortTls(Optional.empty()); + conf.setWebServicePortTls(Optional.empty()); conf.setClientLibraryVersionCheckEnabled(true); startBroker(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientAuthenticationTlsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientAuthenticationTlsTest.java index c9b243257c4..d716d5a8063 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientAuthenticationTlsTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientAuthenticationTlsTest.java @@ -22,6 +22,7 @@ import static org.testng.Assert.assertThrows; import static org.testng.Assert.assertTrue; import static org.testng.Assert.expectThrows; import java.util.HashSet; +import java.util.Optional; import java.util.Set; import java.util.UUID; import java.util.concurrent.TimeUnit; @@ -50,7 +51,8 @@ public class ClientAuthenticationTlsTest extends ProducerConsumerBase { Set<String> providers = new HashSet<>(); providers.add(AuthenticationProviderTls.class.getName()); conf.setAuthenticationProviders(providers); - + conf.setWebServicePortTls(Optional.of(0)); + conf.setBrokerServicePortTls(Optional.of(0)); conf.setTlsKeyFilePath(BROKER_KEY_FILE_PATH); conf.setTlsCertificateFilePath(BROKER_CERT_FILE_PATH); conf.setTlsTrustCertsFilePath(CA_CERT_FILE_PATH); diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyRefreshAuthTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyRefreshAuthTest.java index bde989fc432..2f36cc679f1 100644 --- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyRefreshAuthTest.java +++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyRefreshAuthTest.java @@ -69,6 +69,7 @@ public class ProxyRefreshAuthTest extends ProducerConsumerBase { conf.setAdvertisedAddress(null); conf.setAuthenticateOriginalAuthData(true); conf.setBrokerServicePort(Optional.of(0)); + conf.setWebServicePortTls(Optional.of(0)); conf.setWebServicePort(Optional.of(0)); Set<String> superUserRoles = new HashSet<>(); diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyServiceTlsStarterTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyServiceTlsStarterTest.java index 01c06fbf52f..6247c2a66e8 100644 --- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyServiceTlsStarterTest.java +++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyServiceTlsStarterTest.java @@ -75,6 +75,7 @@ public class ProxyServiceTlsStarterTest extends MockedPulsarServiceBaseTest { protected void doInitConf() throws Exception { super.doInitConf(); + this.conf.setBrokerServicePortTls(Optional.of(0)); this.conf.setTlsCertificateFilePath(PROXY_CERT_FILE_PATH); this.conf.setTlsKeyFilePath(PROXY_KEY_FILE_PATH); } diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/BrokerContainer.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/BrokerContainer.java index 616d45554d7..a51397050b9 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/BrokerContainer.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/BrokerContainer.java @@ -28,8 +28,13 @@ public class BrokerContainer extends PulsarContainer<BrokerContainer> { public static final String NAME = "pulsar-broker"; public BrokerContainer(String clusterName, String hostName) { - super(clusterName, hostName, hostName, "bin/run-broker.sh", BROKER_PORT, BROKER_PORT_TLS, - BROKER_HTTP_PORT, BROKER_HTTPS_PORT, DEFAULT_HTTP_PATH, DEFAULT_IMAGE_NAME); + this(clusterName, hostName, false); + } + + public BrokerContainer(String clusterName, String hostName, boolean enableTls) { + super(clusterName, hostName, hostName, "bin/run-broker.sh", BROKER_PORT, + enableTls ? BROKER_PORT_TLS : 0, BROKER_HTTP_PORT, + enableTls ? BROKER_HTTPS_PORT : 0, DEFAULT_HTTP_PATH, DEFAULT_IMAGE_NAME); tailContainerLog(); } diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/ProxyContainer.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/ProxyContainer.java index 53283447378..f3926878f37 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/ProxyContainer.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/ProxyContainer.java @@ -28,8 +28,13 @@ public class ProxyContainer extends PulsarContainer<ProxyContainer> { public static final String NAME = "pulsar-proxy"; public ProxyContainer(String clusterName, String hostName) { - super(clusterName, hostName, hostName, "bin/run-proxy.sh", BROKER_PORT, BROKER_PORT_TLS, BROKER_HTTP_PORT, - BROKER_HTTPS_PORT, DEFAULT_HTTP_PATH, DEFAULT_IMAGE_NAME); + this(clusterName, hostName, false); + } + + public ProxyContainer(String clusterName, String hostName, boolean enableTls) { + super(clusterName, hostName, hostName, "bin/run-proxy.sh", BROKER_PORT, + enableTls ? BROKER_PORT_TLS : 0, BROKER_HTTP_PORT, + enableTls ? BROKER_HTTPS_PORT : 0, DEFAULT_HTTP_PATH, DEFAULT_IMAGE_NAME); } @Override diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/tls/ClientTlsTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/tls/ClientTlsTest.java index 59ff978cafa..080912cd492 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/tls/ClientTlsTest.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/tls/ClientTlsTest.java @@ -29,6 +29,7 @@ import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.tests.integration.suites.PulsarTestSuite; +import org.apache.pulsar.tests.integration.topologies.PulsarClusterSpec; import org.testng.annotations.DataProvider; import org.testng.annotations.Test; @@ -41,6 +42,14 @@ public class ClientTlsTest extends PulsarTestSuite { return Resources.getResource("certificate-authority/" + name).getPath(); } + @Override + protected PulsarClusterSpec.PulsarClusterSpecBuilder beforeSetupCluster( + String clusterName, + PulsarClusterSpec.PulsarClusterSpecBuilder specBuilder) { + specBuilder.enableTls(true); + return specBuilder; + } + @DataProvider(name = "adminUrls") public Object[][] adminUrls() { return new Object[][]{ diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarCluster.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarCluster.java index 9b4823f46d4..769f135599b 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarCluster.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarCluster.java @@ -38,6 +38,7 @@ import java.util.function.Function; import lombok.Getter; import lombok.extern.slf4j.Slf4j; import org.apache.commons.io.IOUtils; +import org.apache.pulsar.client.impl.auth.AuthenticationTls; import org.apache.pulsar.tests.integration.containers.BKContainer; import org.apache.pulsar.tests.integration.containers.BrokerContainer; import org.apache.pulsar.tests.integration.containers.CSContainer; @@ -132,14 +133,16 @@ public class PulsarCluster { this.brokerContainers = Maps.newTreeMap(); this.workerContainers = Maps.newTreeMap(); - this.proxyContainer = new ProxyContainer(appendClusterName("pulsar-proxy"), ProxyContainer.NAME) + this.proxyContainer = new ProxyContainer(appendClusterName("pulsar-proxy"), ProxyContainer.NAME, spec.enableTls) .withNetwork(network) .withNetworkAliases(appendClusterName("pulsar-proxy")) .withEnv("zkServers", appendClusterName(ZKContainer.NAME)) .withEnv("zookeeperServers", appendClusterName(ZKContainer.NAME)) .withEnv("configurationStoreServers", CSContainer.NAME + ":" + CS_PORT) - .withEnv("clusterName", clusterName) + .withEnv("clusterName", clusterName); // enable mTLS + if (spec.enableTls) { + proxyContainer .withEnv("webServicePortTls", String.valueOf(BROKER_HTTPS_PORT)) .withEnv("servicePortTls", String.valueOf(BROKER_PORT_TLS)) .withEnv("forwardAuthorizationCredentials", "true") @@ -147,7 +150,15 @@ public class PulsarCluster { .withEnv("tlsAllowInsecureConnection", "false") .withEnv("tlsCertificateFilePath", "/pulsar/certificate-authority/server-keys/proxy.cert.pem") .withEnv("tlsKeyFilePath", "/pulsar/certificate-authority/server-keys/proxy.key-pk8.pem") - .withEnv("tlsTrustCertsFilePath", "/pulsar/certificate-authority/certs/ca.cert.pem"); + .withEnv("tlsTrustCertsFilePath", "/pulsar/certificate-authority/certs/ca.cert.pem") + .withEnv("brokerClientAuthenticationPlugin", AuthenticationTls.class.getName()) + .withEnv("brokerClientAuthenticationParameters", String.format("tlsCertFile:%s,tlsKeyFile:%s", "/pulsar/certificate-authority/client-keys/admin.cert.pem", "/pulsar/certificate-authority/client-keys/admin.key-pk8.pem")) + .withEnv("tlsEnabledWithBroker", "true") + .withEnv("brokerClientTrustCertsFilePath", "/pulsar/certificate-authority/certs/ca.cert.pem") + .withEnv("brokerClientCertificateFilePath", "/pulsar/certificate-authority/server-keys/proxy.cert.pem") + .withEnv("brokerClientKeyFilePath", "/pulsar/certificate-authority/server-keys/proxy.key-pk8.pem"); + + } if (spec.proxyEnvs != null) { spec.proxyEnvs.forEach(this.proxyContainer::withEnv); } @@ -184,7 +195,7 @@ public class PulsarCluster { // create brokers brokerContainers.putAll( runNumContainers("broker", spec.numBrokers(), (name) -> { - BrokerContainer brokerContainer = new BrokerContainer(clusterName, appendClusterName(name)) + BrokerContainer brokerContainer = new BrokerContainer(clusterName, appendClusterName(name), spec.enableTls) .withNetwork(network) .withNetworkAliases(appendClusterName(name)) .withEnv("zkServers", appendClusterName(ZKContainer.NAME)) @@ -195,16 +206,19 @@ public class PulsarCluster { .withEnv("loadBalancerOverrideBrokerNicSpeedGbps", "1") // used in s3 tests .withEnv("AWS_ACCESS_KEY_ID", "accesskey").withEnv("AWS_SECRET_KEY", "secretkey") - .withEnv("maxMessageSize", "" + spec.maxMessageSize) + .withEnv("maxMessageSize", "" + spec.maxMessageSize); + if (spec.enableTls) { // enable mTLS - .withEnv("webServicePortTls", String.valueOf(BROKER_HTTPS_PORT)) - .withEnv("brokerServicePortTls", String.valueOf(BROKER_PORT_TLS)) - .withEnv("authenticateOriginalAuthData", "true") - .withEnv("tlsRequireTrustedClientCertOnConnect", "true") - .withEnv("tlsAllowInsecureConnection", "false") - .withEnv("tlsCertificateFilePath", "/pulsar/certificate-authority/server-keys/broker.cert.pem") - .withEnv("tlsKeyFilePath", "/pulsar/certificate-authority/server-keys/broker.key-pk8.pem") - .withEnv("tlsTrustCertsFilePath", "/pulsar/certificate-authority/certs/ca.cert.pem"); + brokerContainer + .withEnv("webServicePortTls", String.valueOf(BROKER_HTTPS_PORT)) + .withEnv("brokerServicePortTls", String.valueOf(BROKER_PORT_TLS)) + .withEnv("authenticateOriginalAuthData", "true") + .withEnv("tlsAllowInsecureConnection", "false") + .withEnv("tlsRequireTrustedClientCertOnConnect", "true") + .withEnv("tlsTrustCertsFilePath", "/pulsar/certificate-authority/certs/ca.cert.pem") + .withEnv("tlsCertificateFilePath", "/pulsar/certificate-authority/server-keys/broker.cert.pem") + .withEnv("tlsKeyFilePath", "/pulsar/certificate-authority/server-keys/broker.key-pk8.pem"); + } if (spec.queryLastMessage) { brokerContainer.withEnv("bookkeeperExplicitLacIntervalInMills", "10"); brokerContainer.withEnv("bookkeeperUseV2WireProtocol", "false"); diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarClusterSpec.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarClusterSpec.java index fa28d20e6b3..c141e990d62 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarClusterSpec.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarClusterSpec.java @@ -177,4 +177,10 @@ public class PulsarClusterSpec { * Additional ports to expose on bookie containers. */ List<Integer> bookieAdditionalPorts; + + /** + * Enable TLS for connection. + */ + @Default + boolean enableTls = false; }
