pulsar-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From GitBox <...@apache.org>
Subject [GitHub] sijie closed pull request #2988: Start Pulsar in TLS Only mode and deprecate tlsEnabled flag.
Date Sun, 16 Dec 2018 01:37:04 GMT
sijie closed pull request #2988: Start Pulsar in TLS Only mode and deprecate tlsEnabled flag.
URL: https://github.com/apache/pulsar/pull/2988
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git a/.gitignore b/.gitignore
index e43e519a71..f605f1e8ac 100644
--- a/.gitignore
+++ b/.gitignore
@@ -20,6 +20,7 @@ pulsar-broker/src/test/resources/pulsar-functions-api-examples.jar
 .project
 .settings/
 .recommenders/
+.factorypath
 
 # Intellij
 .idea/
diff --git a/conf/broker.conf b/conf/broker.conf
index 3c1853adc5..199acd3692 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -28,14 +28,14 @@ configurationStoreServers=
 # Broker data port
 brokerServicePort=6650
 
-# Broker data port for TLS
-brokerServicePortTls=6651
+# Broker data port for TLS - By default TLS is disabled
+brokerServicePortTls=
 
 # Port to use to server HTTP request
 webServicePort=8080
 
-# Port to use to server HTTPS request
-webServicePortTls=8443
+# Port to use to server HTTPS request - By default TLS is disabled
+webServicePortTls=
 
 # Hostname or IP address the service binds on, default is 0.0.0.0.
 bindAddress=0.0.0.0
@@ -220,7 +220,7 @@ proxyRoles=
 # else it just accepts the originalPrincipal and authorizes it (if required).
 authenticateOriginalAuthData=false
 
-# Enable TLS
+# Deprecated - Use webServicePortTls and brokerServicePortTls instead
 tlsEnabled=false
 
 # Path for the TLS certificate file
diff --git a/conf/discovery.conf b/conf/discovery.conf
index b1b6f41017..907f546378 100644
--- a/conf/discovery.conf
+++ b/conf/discovery.conf
@@ -30,13 +30,13 @@ zookeeperSessionTimeoutMs=30000
 servicePort=6650
 
 # Port to use to server binary-proto-tls request
-servicePortTls=6651
+servicePortTls=
 
 # Port that discovery service listen on
 webServicePort=8080
 
 # Port to use to server HTTPS request
-webServicePortTls=8443
+webServicePortTls=
 
 # Control whether to bind directly on localhost rather than on normal hostname
 bindOnLocalhost=false
@@ -65,7 +65,7 @@ superUserRoles=
 authorizationAllowWildcardsMatching=false
 
 ##### --- TLS --- #####
-# Enable TLS
+# Deprecated - Use servicePortTls and webServicePortTls instead
 tlsEnabled=false
 
 # Path for the TLS certificate file
diff --git a/conf/proxy.conf b/conf/proxy.conf
index ffa6c45fad..78dbe00390 100644
--- a/conf/proxy.conf
+++ b/conf/proxy.conf
@@ -43,13 +43,13 @@ zookeeperSessionTimeoutMs=30000
 servicePort=6650
 
 # The port to use to server binary Protobuf TLS requests
-servicePortTls=6651
+servicePortTls=
 
 # Port that discovery service listen on
 webServicePort=8080
 
 # Port to use to server HTTPS request
-webServicePortTls=8443
+webServicePortTls=
 
 # Path for the file used to determine the rotation status for the proxy instance when responding
 # to service discovery health checks
@@ -98,7 +98,7 @@ maxConcurrentLookupRequests=50000
 
 ##### --- TLS --- #####
 
-# Whether TLS is enabled for the proxy
+# Deprecated - use servicePortTls and webServicePortTls instead 
 tlsEnabledInProxy=false
 
 # Whether TLS is enabled when communicating with Pulsar brokers
diff --git a/conf/websocket.conf b/conf/websocket.conf
index c8144fd3be..ab03e5687d 100644
--- a/conf/websocket.conf
+++ b/conf/websocket.conf
@@ -34,7 +34,7 @@ brokerServiceUrlTls=
 # Port to use to server HTTP request
 webServicePort=8080
 # Port to use to server HTTPS request
-webServicePortTls=8443
+webServicePortTls=
 
 # Path for the file used to determine the rotation status for the proxy-instance when responding
 # to service discovery health checks
@@ -79,6 +79,7 @@ authorizationAllowWildcardsMatching=false
 superUserRoles=
 
 # Authentication settings of the proxy itself. Used to connect to brokers
+brokerClientTlsEnabled=false;
 brokerClientAuthenticationPlugin=
 brokerClientAuthenticationParameters=
 brokerClientTrustCertsFilePath=
@@ -88,7 +89,7 @@ anonymousUserRole=
 
 ### --- TLS --- ###
 
-# Enable TLS
+# Deprecated - use webServicePortTls and brokerClientTlsEnabled instead 
 tlsEnabled=false
 
 # Accept untrusted TLS certificate from client
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index 34882bd16c..61a8775dd6 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -52,12 +52,12 @@
     // Configuration Store connection string
     @FieldContext(required = false)
     private String configurationStoreServers;
-    private int brokerServicePort = 6650;
-    private int brokerServicePortTls = 6651;
+    private Integer brokerServicePort = 6650;
+    private Integer brokerServicePortTls = null;
     // Port to use to server HTTP request
-    private int webServicePort = 8080;
+    private Integer webServicePort = 8080;
     // Port to use to server HTTPS request
-    private int webServicePortTls = 8443;
+    private Integer webServicePortTls = null;
 
     // Hostname or IP address the service binds on.
     private String bindAddress = "0.0.0.0";
@@ -228,7 +228,7 @@
     private int maxConsumersPerSubscription = 0;
 
     /***** --- TLS --- ****/
-    // Enable TLS
+    @Deprecated
     private boolean tlsEnabled = false;
     // Path for the TLS certificate file
     private String tlsCertificateFilePath;
@@ -584,5 +584,20 @@ public void setProperties(Properties properties) {
     public int getBookkeeperHealthCheckIntervalSec() {
         return (int) bookkeeperClientHealthCheckIntervalSeconds;
     }
+    
+    public Optional<Integer> getBrokerServicePort() {
+        return Optional.ofNullable(brokerServicePort);
+    }
+    
+    public Optional<Integer> getBrokerServicePortTls() {
+        return Optional.ofNullable(brokerServicePortTls);
+    }
+    
+    public Optional<Integer> getWebServicePort() {
+        return Optional.ofNullable(webServicePort);
+    }
 
+    public Optional<Integer> getWebServicePortTls() {
+        return Optional.ofNullable(webServicePortTls);
+    }
 }
diff --git a/pulsar-broker-common/src/test/java/org/apache/pulsar/common/configuration/PulsarConfigurationLoaderTest.java b/pulsar-broker-common/src/test/java/org/apache/pulsar/common/configuration/PulsarConfigurationLoaderTest.java
index baa0e7558c..34590863a7 100644
--- a/pulsar-broker-common/src/test/java/org/apache/pulsar/common/configuration/PulsarConfigurationLoaderTest.java
+++ b/pulsar-broker-common/src/test/java/org/apache/pulsar/common/configuration/PulsarConfigurationLoaderTest.java
@@ -67,10 +67,10 @@ public void testConfigurationConverting() throws Exception {
         // check whether converting correctly
         assertEquals(serviceConfiguration.getZookeeperServers(), "localhost:2181");
         assertEquals(serviceConfiguration.getConfigurationStoreServers(), "localhost:2184");
-        assertEquals(serviceConfiguration.getBrokerServicePort(), 7650);
-        assertEquals(serviceConfiguration.getBrokerServicePortTls(), 7651);
-        assertEquals(serviceConfiguration.getWebServicePort(), 9080);
-        assertEquals(serviceConfiguration.getWebServicePortTls(), 9443);
+        assertEquals(serviceConfiguration.getBrokerServicePort().get(), new Integer(7650));
+        assertEquals(serviceConfiguration.getBrokerServicePortTls().get(), new Integer(7651));
+        assertEquals(serviceConfiguration.getWebServicePort().get(), new Integer(9080));
+        assertEquals(serviceConfiguration.getWebServicePortTls().get(), new Integer(9443));
 
         // check whether exception causes
         try {
@@ -112,7 +112,7 @@ public void testPulsarConfiguraitonLoadingStream() throws Exception {
         assertEquals(serviceConfig.getBacklogQuotaDefaultLimitGB(), 18);
         assertEquals(serviceConfig.getClusterName(), "usc");
         assertEquals(serviceConfig.getBrokerClientAuthenticationParameters(), "role:my-role");
-        assertEquals(serviceConfig.getBrokerServicePort(), 7777);
+        assertEquals(serviceConfig.getBrokerServicePort().get(), new Integer(7777));
         assertEquals(serviceConfig.getManagedLedgerDigestType(), DigestType.CRC32C);
     }
 
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarBrokerStarter.java b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarBrokerStarter.java
index 0af322facb..1cf1b5f888 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarBrokerStarter.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarBrokerStarter.java
@@ -147,17 +147,17 @@ private static boolean argsContains(String[] args, String arg) {
                 boolean useTls = workerConfig.isUseTls();
                 String localhost = "127.0.0.1";
                 String pulsarServiceUrl = useTls
-                        ? PulsarService.brokerUrlTls(localhost, brokerConfig.getBrokerServicePortTls())
-                        : PulsarService.brokerUrl(localhost, brokerConfig.getBrokerServicePort());
+                        ? PulsarService.brokerUrlTls(localhost, brokerConfig.getBrokerServicePortTls().get())
+                        : PulsarService.brokerUrl(localhost, brokerConfig.getBrokerServicePort().get());
                 String webServiceUrl = useTls
-                        ? PulsarService.webAddressTls(localhost, brokerConfig.getWebServicePortTls())
-                        : PulsarService.webAddress(localhost, brokerConfig.getWebServicePort());
+                        ? PulsarService.webAddressTls(localhost, brokerConfig.getWebServicePortTls().get())
+                        : PulsarService.webAddress(localhost, brokerConfig.getWebServicePort().get());
                 workerConfig.setPulsarServiceUrl(pulsarServiceUrl);
                 workerConfig.setPulsarWebServiceUrl(webServiceUrl);
                 String hostname = ServiceConfigurationUtils.getDefaultOrConfiguredAddress(
                     brokerConfig.getAdvertisedAddress());
                 workerConfig.setWorkerHostname(hostname);
-                workerConfig.setWorkerPort(brokerConfig.getWebServicePort());
+                workerConfig.setWorkerPort(brokerConfig.getWebServicePort().get());
                 workerConfig.setWorkerId(
                     "c-" + brokerConfig.getClusterName()
                         + "-fw-" + hostname
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarStandalone.java b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarStandalone.java
index d1f37c9dbd..8711bd84d6 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarStandalone.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarStandalone.java
@@ -272,8 +272,8 @@ void start() throws Exception {
                 workerConfig = WorkerConfig.load(this.getFnWorkerConfigFile());
             }
             // worker talks to local broker
-            workerConfig.setPulsarServiceUrl("pulsar://127.0.0.1:" + config.getBrokerServicePort());
-            workerConfig.setPulsarWebServiceUrl("http://127.0.0.1:" + config.getWebServicePort());
+            workerConfig.setPulsarServiceUrl("pulsar://127.0.0.1:" + config.getBrokerServicePort().get());
+            workerConfig.setPulsarWebServiceUrl("http://127.0.0.1:" + config.getWebServicePort().get());
             if (!this.isNoStreamStorage()) {
                 // only set the state storage service url when state is enabled.
                 workerConfig.setStateStorageServiceUrl("bk://127.0.0.1:" + this.getStreamStoragePort());
@@ -281,7 +281,7 @@ void start() throws Exception {
             String hostname = ServiceConfigurationUtils.getDefaultOrConfiguredAddress(
                 config.getAdvertisedAddress());
             workerConfig.setWorkerHostname(hostname);
-            workerConfig.setWorkerPort(config.getWebServicePort());
+            workerConfig.setWorkerPort(config.getWebServicePort().get());
             workerConfig.setWorkerId(
                 "c-" + config.getClusterName()
                     + "-fw-" + hostname
@@ -294,9 +294,9 @@ void start() throws Exception {
         broker.start();
 
         URL webServiceUrl = new URL(
-                String.format("http://%s:%d", config.getAdvertisedAddress(), config.getWebServicePort()));
+                String.format("http://%s:%d", config.getAdvertisedAddress(), config.getWebServicePort().get()));
         final String brokerServiceUrl = String.format("pulsar://%s:%d", config.getAdvertisedAddress(),
-                config.getBrokerServicePort());
+                config.getBrokerServicePort().get());
         admin = PulsarAdmin.builder().serviceHttpUrl(webServiceUrl.toString()).authentication(
                 config.getBrokerClientAuthenticationPlugin(), config.getBrokerClientAuthenticationParameters()).build();
 
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/MessagingServiceShutdownHook.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/MessagingServiceShutdownHook.java
index 1f1007eec7..6cacaa53de 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/MessagingServiceShutdownHook.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/MessagingServiceShutdownHook.java
@@ -49,7 +49,7 @@ public MessagingServiceShutdownHook(PulsarService service) {
     public void run() {
         if (service.getConfiguration() != null) {
             LOG.info("messaging service shutdown hook started, lookup port="
-                    + service.getConfiguration().getWebServicePort() + ", broker url=" + service.getBrokerServiceUrl());
+                    + service.getConfiguration().getWebServicePort().get() + ", broker url=" + service.getBrokerServiceUrl());
         }
 
         ExecutorService executor = Executors.newSingleThreadExecutor(new DefaultThreadFactory("shutdown-thread"));
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 4f5bd7288f..74e91bee9f 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
@@ -470,7 +470,7 @@ public synchronized void brokerIsAFollowerNow() {
             this.startWorkerService();
 
             LOG.info("messaging service is ready, bootstrap service on port={}, broker url={}, cluster={}, configs={}",
-                    config.getWebServicePort(), brokerServiceUrl, config.getClusterName(),
+                    config.getWebServicePort().get(), brokerServiceUrl, config.getClusterName(),
                     ReflectionToStringBuilder.toString(config));
         } catch (Exception e) {
             LOG.error(e.getMessage(), e);
@@ -855,7 +855,11 @@ public static String advertisedAddress(ServiceConfiguration config) {
     }
 
     public static String brokerUrl(ServiceConfiguration config) {
-        return brokerUrl(advertisedAddress(config), config.getBrokerServicePort());
+        if (config.getBrokerServicePort().isPresent()) {
+            return brokerUrl(advertisedAddress(config), config.getBrokerServicePort().get());
+        } else {
+            return null;
+        }
     }
 
     public static String brokerUrl(String host, int port) {
@@ -863,10 +867,10 @@ public static String brokerUrl(String host, int port) {
     }
 
     public static String brokerUrlTls(ServiceConfiguration config) {
-        if (config.isTlsEnabled()) {
-            return brokerUrlTls(advertisedAddress(config), config.getBrokerServicePortTls());
+        if (config.getBrokerServicePortTls().isPresent()) {
+            return brokerUrlTls(advertisedAddress(config), config.getBrokerServicePortTls().get());
         } else {
-            return "";
+            return null;
         }
     }
 
@@ -875,7 +879,11 @@ public static String brokerUrlTls(String host, int port) {
     }
 
     public static String webAddress(ServiceConfiguration config) {
-        return webAddress(advertisedAddress(config), config.getWebServicePort());
+        if (config.getWebServicePort().isPresent()) {        
+            return webAddress(advertisedAddress(config), config.getWebServicePort().get());
+        } else {
+            return null;
+        }
     }
 
     public static String webAddress(String host, int port) {
@@ -883,10 +891,10 @@ public static String webAddress(String host, int port) {
     }
 
     public static String webAddressTls(ServiceConfiguration config) {
-        if (config.isTlsEnabled()) {
-            return webAddressTls(advertisedAddress(config), config.getWebServicePortTls());
+        if (config.getWebServicePortTls().isPresent()) {
+            return webAddressTls(advertisedAddress(config), config.getWebServicePortTls().get());
         } else {
-            return "";
+            return null;
         }
     }
 
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
index 79e21d4b90..fed0004ee0 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
@@ -289,8 +289,8 @@ protected void validateTopicName(String property, String cluster, String namespa
     protected void validateBrokerName(String broker) throws MalformedURLException {
         String brokerUrl = String.format("http://%s", broker);
         String brokerUrlTls = String.format("https://%s", broker);
-        if (!pulsar().getWebServiceAddress().equals(brokerUrl)
-                && !pulsar().getWebServiceAddressTls().equals(brokerUrlTls)) {
+        if (!brokerUrl.equals(pulsar().getWebServiceAddress())
+                && !brokerUrlTls.equals(pulsar().getWebServiceAddressTls())) {
             String[] parts = broker.split(":");
             checkArgument(parts.length == 2, "Invalid broker url %s", broker);
             String host = parts[0];
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/NoopLoadManager.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/NoopLoadManager.java
index 5773c61ee1..087eb2804d 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/NoopLoadManager.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/NoopLoadManager.java
@@ -54,7 +54,7 @@
 
     @Override
     public void initialize(PulsarService pulsar) {
-        lookupServiceAddress = pulsar.getAdvertisedAddress() + ":" + pulsar.getConfiguration().getWebServicePort();
+        lookupServiceAddress = pulsar.getAdvertisedAddress() + ":" + pulsar.getConfiguration().getWebServicePort().get();
         localResourceUnit = new SimpleResourceUnit(String.format("http://%s", lookupServiceAddress),
                 new PulsarResourceDescription());
         zkClient = pulsar.getZkClient();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
index 45a31f5c47..997d9bb9d9 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
@@ -783,7 +783,7 @@ public void start() throws PulsarServerException {
             // Register the brokers in zk list
             createZPathIfNotExists(zkClient, LoadManager.LOADBALANCE_BROKERS_ROOT);
 
-            String lookupServiceAddress = pulsar.getAdvertisedAddress() + ":" + conf.getWebServicePort();
+            String lookupServiceAddress = pulsar.getAdvertisedAddress() + ":" + conf.getWebServicePort().get();
             brokerZnodePath = LoadManager.LOADBALANCE_BROKERS_ROOT + "/" + lookupServiceAddress;
             final String timeAverageZPath = TIME_AVERAGE_BROKER_ZPATH + "/" + lookupServiceAddress;
             updateLocalBrokerData();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/SimpleLoadManagerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/SimpleLoadManagerImpl.java
index da4c0e3e6d..023ce3b1a3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/SimpleLoadManagerImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/SimpleLoadManagerImpl.java
@@ -283,7 +283,7 @@ public void start() throws PulsarServerException {
                     // ignore the exception, node might be present already
                 }
             }
-            String lookupServiceAddress = pulsar.getAdvertisedAddress() + ":" + conf.getWebServicePort();
+            String lookupServiceAddress = pulsar.getAdvertisedAddress() + ":" + conf.getWebServicePort().get();
             brokerZnodePath = LOADBALANCE_BROKERS_ROOT + "/" + lookupServiceAddress;
             LoadReport loadReport = null;
             try {
@@ -1113,7 +1113,7 @@ private LoadReport generateLoadReportForcefully() throws Exception {
                 loadReport.setNonPersistentTopicsEnabled(pulsar.getConfiguration().isEnableNonPersistentTopics());
                 loadReport.setPersistentTopicsEnabled(pulsar.getConfiguration().isEnablePersistentTopics());
                 loadReport.setName(String.format("%s:%s", pulsar.getAdvertisedAddress(),
-                        pulsar.getConfiguration().getWebServicePort()));
+                        pulsar.getConfiguration().getWebServicePort().get()));
                 loadReport.setBrokerVersionString(pulsar.getBrokerVersion());
 
                 SystemResourceUsage systemResourceUsage = this.getSystemResourceUsage();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
index ab8aef7368..9b4fcaef39 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
@@ -487,7 +487,7 @@ private boolean isBrokerActive(String candidateBroker) throws KeeperException, I
 
         if (LOG.isDebugEnabled()) {
             LOG.debug("Broker not found for SLA Monitoring Namespace {}",
-                    candidateBroker + ":" + config.getWebServicePort());
+                    candidateBroker + ":" + config.getWebServicePort().get());
         }
         return false;
     }
@@ -972,11 +972,22 @@ public void unloadSLANamespace() throws Exception {
     }
 
     public static String getHeartbeatNamespace(String host, ServiceConfiguration config) {
-        return String.format(HEARTBEAT_NAMESPACE_FMT, config.getClusterName(), host, config.getWebServicePort());
+        Integer port = null;
+        if (config.getWebServicePort().isPresent()) {
+            port = config.getWebServicePort().get();
+        } else if (config.getWebServicePortTls().isPresent()) {
+            port = config.getWebServicePortTls().get();
+        } 
+        return String.format(HEARTBEAT_NAMESPACE_FMT, config.getClusterName(), host, port);
     }
-
-    public static String getSLAMonitorNamespace(String host, ServiceConfiguration config) {
-        return String.format(SLA_NAMESPACE_FMT, config.getClusterName(), host, config.getWebServicePort());
+     public static String getSLAMonitorNamespace(String host, ServiceConfiguration config) {
+        Integer port = null;
+        if (config.getWebServicePort().isPresent()) {
+            port = config.getWebServicePort().get();
+        } else if (config.getWebServicePortTls().isPresent()) {
+            port = config.getWebServicePortTls().get();
+        } 
+        return String.format(SLA_NAMESPACE_FMT, config.getClusterName(), host, port);
     }
 
     public static String checkHeartbeatNamespace(ServiceUnitId ns) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index 16352a4f49..a43e404dba 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -135,8 +135,6 @@
 
     private final PulsarService pulsar;
     private final ManagedLedgerFactory managedLedgerFactory;
-    private final int port;
-    private final int tlsPort;
 
     private final ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics;
 
@@ -192,8 +190,6 @@
     public BrokerService(PulsarService pulsar) throws Exception {
         this.pulsar = pulsar;
         this.managedLedgerFactory = pulsar.getManagedLedgerFactory();
-        this.port = new URI(pulsar.getBrokerServiceUrl()).getPort();
-        this.tlsPort = new URI(pulsar.getBrokerServiceUrlTls()).getPort();
         this.topics = new ConcurrentOpenHashMap<>();
         this.replicationClients = new ConcurrentOpenHashMap<>();
         this.keepAliveIntervalSeconds = pulsar.getConfiguration().getKeepAliveIntervalSeconds();
@@ -293,20 +289,24 @@ public void start() throws Exception {
 
         bootstrap.childHandler(new PulsarChannelInitializer(pulsar, false));
 
-        // Bind and start to accept incoming connections.
-        InetSocketAddress addr = new InetSocketAddress(pulsar.getBindAddress(), port);
-        try {
-            bootstrap.bind(addr).sync();
-        } catch (Exception e) {
-            throw new IOException("Failed to bind Pulsar broker on " + addr, e);
+        Optional<Integer> port = serviceConfig.getBrokerServicePort();
+        if (port.isPresent()) {
+            // Bind and start to accept incoming connections.
+            InetSocketAddress addr = new InetSocketAddress(pulsar.getBindAddress(), port.get());
+            try {
+                bootstrap.bind(addr).sync();
+            } catch (Exception e) {
+                throw new IOException("Failed to bind Pulsar broker on " + addr, e);
+            }
+            log.info("Started Pulsar Broker service on port {}", port.get());
         }
-        log.info("Started Pulsar Broker service on port {}", port);
-
-        if (serviceConfig.isTlsEnabled()) {
+        
+        Optional<Integer> tlsPort = serviceConfig.getBrokerServicePortTls();
+        if (tlsPort.isPresent()) {
             ServerBootstrap tlsBootstrap = bootstrap.clone();
             tlsBootstrap.childHandler(new PulsarChannelInitializer(pulsar, true));
-            tlsBootstrap.bind(new InetSocketAddress(pulsar.getBindAddress(), tlsPort)).sync();
-            log.info("Started Pulsar Broker TLS service on port {} - TLS provider: {}", tlsPort,
+            tlsBootstrap.bind(new InetSocketAddress(pulsar.getBindAddress(), tlsPort.get())).sync();
+            log.info("Started Pulsar Broker TLS service on port {} - TLS provider: {}", tlsPort.get(),
                     SslContext.defaultServerProvider());
         }
 
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
index a9557116e5..81e5a40a31 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
@@ -320,7 +320,7 @@ protected void validateClusterOwnership(String cluster) throws WebApplicationExc
 
     private URI getRedirectionUrl(ClusterData differentClusterData) throws MalformedURLException {
         URL webUrl = null;
-        if (isRequestHttps() && pulsar.getConfiguration().isTlsEnabled()
+        if (isRequestHttps() && pulsar.getConfiguration().getWebServicePortTls().isPresent()
                 && StringUtils.isNotBlank(differentClusterData.getServiceUrlTls())) {
             webUrl = new URL(differentClusterData.getServiceUrlTls());
         } else {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java
index 02243b9558..4ca8fcf70a 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java
@@ -27,6 +27,7 @@
 import java.util.EnumSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.TimeZone;
 
 import javax.servlet.DispatcherType;
@@ -81,12 +82,16 @@ public WebService(PulsarService pulsar) throws PulsarServerException {
         this.server = new Server(webServiceExecutor);
         List<ServerConnector> connectors = new ArrayList<>();
 
-        ServerConnector connector = new PulsarServerConnector(server, 1, 1);
-        connector.setPort(pulsar.getConfiguration().getWebServicePort());
-        connector.setHost(pulsar.getBindAddress());
-        connectors.add(connector);
+        Optional<Integer> port = pulsar.getConfiguration().getWebServicePort();
+        if (port.isPresent()) {
+            ServerConnector connector = new PulsarServerConnector(server, 1, 1);
+            connector.setPort(port.get());
+            connector.setHost(pulsar.getBindAddress());
+            connectors.add(connector);
+        }
 
-        if (pulsar.getConfiguration().isTlsEnabled()) {
+        Optional<Integer> tlsPort = pulsar.getConfiguration().getWebServicePortTls();
+        if (tlsPort.isPresent()) {
             try {
                 SslContextFactory sslCtxFactory = SecurityUtility.createSslContextFactory(
                         pulsar.getConfiguration().isTlsAllowInsecureConnection(),
@@ -95,7 +100,7 @@ public WebService(PulsarService pulsar) throws PulsarServerException {
                         pulsar.getConfiguration().getTlsKeyFilePath(),
                         pulsar.getConfiguration().isTlsRequireTrustedClientCertOnConnect());
                 ServerConnector tlsConnector = new PulsarServerConnector(server, 1, 1, sslCtxFactory);
-                tlsConnector.setPort(pulsar.getConfiguration().getWebServicePortTls());
+                tlsConnector.setPort(tlsPort.get());
                 tlsConnector.setHost(pulsar.getBindAddress());
                 connectors.add(tlsConnector);
             } catch (GeneralSecurityException e) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorTool.java b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorTool.java
index c16d6767cd..081662534b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorTool.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorTool.java
@@ -79,18 +79,23 @@ public static void main(String[] args) throws Exception {
                     arguments.brokerConfigFile, ServiceConfiguration.class);
         }
 
-        String pulsarServiceUrl = PulsarService.brokerUrl(brokerConfig);
         ClientBuilder clientBuilder = PulsarClient.builder();
 
         if (isNotBlank(brokerConfig.getBrokerClientAuthenticationPlugin())) {
             clientBuilder.authentication(brokerConfig.getBrokerClientAuthenticationPlugin(),
                     brokerConfig.getBrokerClientAuthenticationParameters());
         }
-        clientBuilder.serviceUrl(pulsarServiceUrl)
-                .enableTls(brokerConfig.isTlsEnabled())
-                .allowTlsInsecureConnection(brokerConfig.isTlsAllowInsecureConnection())
-                .tlsTrustCertsFilePath(brokerConfig.getTlsCertificateFilePath());
+        
 
+        if (brokerConfig.getBrokerServicePortTls().isPresent()) {
+            clientBuilder.serviceUrl(PulsarService.brokerUrlTls(brokerConfig))
+            .allowTlsInsecureConnection(brokerConfig.isTlsAllowInsecureConnection())
+            .tlsTrustCertsFilePath(brokerConfig.getTlsCertificateFilePath());
+
+        } else {
+            clientBuilder.serviceUrl(PulsarService.brokerUrl(brokerConfig));
+        }
+        
         ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(
                 new ThreadFactoryBuilder().setNameFormat("compaction-%d").setDaemon(true).build());
 
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/SLAMonitoringTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/SLAMonitoringTest.java
index 233d6b7b36..cddf3abb55 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/SLAMonitoringTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/SLAMonitoringTest.java
@@ -217,8 +217,8 @@ public void testUnloadIfBrokerCrashes() {
         try {
             pulsarServices[crashIndex] = new PulsarService(configurations[crashIndex]);
             pulsarServices[crashIndex].start();
-            assertEquals(pulsarServices[crashIndex].getConfiguration().getBrokerServicePort(),
-                    brokerNativeBrokerPorts[crashIndex]);
+            assertEquals(pulsarServices[crashIndex].getConfiguration().getBrokerServicePort().get(),
+                    new Integer(brokerNativeBrokerPorts[crashIndex]));
         } catch (PulsarServerException e) {
             e.printStackTrace();
             fail("The broker should be able to start without exception");
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 01160fec8d..f7c7e70359 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
@@ -139,7 +139,6 @@
     @Override
     public void setup() throws Exception {
         conf.setLoadBalancerEnabled(true);
-        conf.setTlsEnabled(true);
         conf.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
         conf.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
 
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java
index 0b04ba3c5f..3f90392063 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java
@@ -835,7 +835,7 @@ public void brokerNamespaceIsolationPolicies() throws Exception {
         String namespaceRegex = "other/use/other.*";
         String cluster = "use";
         String brokerName = pulsar.getAdvertisedAddress();
-        String brokerAddress = brokerName + ":" + pulsar.getConfiguration().getWebServicePort();
+        String brokerAddress = brokerName + ":" + pulsar.getConfiguration().getWebServicePort().get();
         NamespaceIsolationData nsPolicyData1 = new NamespaceIsolationData();
         nsPolicyData1.namespaces = new ArrayList<String>();
         nsPolicyData1.namespaces.add(namespaceRegex);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTlsAuthTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTlsAuthTest.java
index ca1817a053..b4e2a919ab 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTlsAuthTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTlsAuthTest.java
@@ -61,7 +61,6 @@ private static String getTLSFile(String name) {
     @Override
     public void setup() throws Exception {
         conf.setLoadBalancerEnabled(true);
-        conf.setTlsEnabled(true);
         conf.setTlsCertificateFilePath(getTLSFile("broker.cert"));
         conf.setTlsKeyFilePath(getTLSFile("broker.key-pk8"));
         conf.setTlsTrustCertsFilePath(getTLSFile("ca.cert"));
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/BrokerAdminClientTlsAuthTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/BrokerAdminClientTlsAuthTest.java
index 2cfb009ac7..e478631513 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/BrokerAdminClientTlsAuthTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/BrokerAdminClientTlsAuthTest.java
@@ -83,7 +83,6 @@ public void setup() throws Exception {
 
     private void buildConf(ServiceConfiguration conf) {
         conf.setLoadBalancerEnabled(true);
-        conf.setTlsEnabled(true);
         conf.setTlsCertificateFilePath(getTLSFile("broker.cert"));
         conf.setTlsKeyFilePath(getTLSFile("broker.key-pk8"));
         conf.setTlsTrustCertsFilePath(getTLSFile("ca.cert"));
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v1/V1_AdminApiTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v1/V1_AdminApiTest.java
index 803f8af028..8da7698d88 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v1/V1_AdminApiTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v1/V1_AdminApiTest.java
@@ -138,7 +138,6 @@
     @Override
     public void setup() throws Exception {
         conf.setLoadBalancerEnabled(true);
-        conf.setTlsEnabled(true);
         conf.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
         conf.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
 
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/AntiAffinityNamespaceGroupTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/AntiAffinityNamespaceGroupTest.java
index 8088637855..befc46da0d 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/AntiAffinityNamespaceGroupTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/AntiAffinityNamespaceGroupTest.java
@@ -174,7 +174,7 @@ void shutdown() throws Exception {
     private void createCluster(ZooKeeper zk, ServiceConfiguration config) throws Exception {
         ZkUtils.createFullPathOptimistic(zk, "/admin/clusters/" + config.getClusterName(),
                 ObjectMapperFactory.getThreadLocal().writeValueAsBytes(
-                        new ClusterData("http://" + config.getAdvertisedAddress() + ":" + config.getWebServicePort())),
+                        new ClusterData("http://" + config.getAdvertisedAddress() + ":" + config.getWebServicePort().get())),
                 Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
     }
 
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 b90fa6faa2..a959dee75a 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
@@ -148,7 +148,7 @@ void setup() throws Exception {
             pulsarServices[i].start();
 
             brokerUrls[i] = new URL("http://127.0.0.1" + ":" + brokerWebServicePorts[i]);
-            lookupAddresses[i] = pulsarServices[i].getAdvertisedAddress() + ":" + config.getWebServicePort();
+            lookupAddresses[i] = pulsarServices[i].getAdvertisedAddress() + ":" + config.getWebServicePort().get();
             pulsarAdmins[i] = PulsarAdmin.builder().serviceHttpUrl(brokerUrls[i].toString()).build();
         }
 
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 78b591d2e8..aaa60794e7 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
@@ -251,7 +251,7 @@ public void testPrimary() throws Exception {
         rd.put("bandwidthOut", new ResourceUsage(550 * 1024, 1024 * 1024));
 
         ResourceUnit ru1 = new SimpleResourceUnit(
-                "http://" + pulsar1.getAdvertisedAddress() + ":" + pulsar1.getConfiguration().getWebServicePort(), rd);
+                "http://" + pulsar1.getAdvertisedAddress() + ":" + pulsar1.getConfiguration().getWebServicePort().get(), rd);
         Set<ResourceUnit> rus = new HashSet<ResourceUnit>();
         rus.add(ru1);
         LoadRanker lr = new ResourceAvailabilityRanker();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnershipCacheTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnershipCacheTest.java
index dfca1b2c06..1c11f00b0d 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnershipCacheTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnershipCacheTest.java
@@ -93,7 +93,8 @@ public void setup() throws Exception {
         doReturn(localCache).when(pulsar).getLocalZkCacheService();
         doReturn(config).when(pulsar).getConfiguration();
         doReturn(nsService).when(pulsar).getNamespaceService();
-        doReturn(port).when(config).getBrokerServicePort();
+        doReturn(Optional.ofNullable(new Integer(port))).when(config).getBrokerServicePort();
+        doReturn(Optional.ofNullable(null)).when(config).getWebServicePort();
         doReturn(brokerService).when(pulsar).getBrokerService();
         doReturn(webAddress(config)).when(pulsar).getWebServiceAddress();
         doReturn(selfBrokerUrl).when(pulsar).getBrokerServiceUrl();
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 54d1644975..4e9f4e986f 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
@@ -408,7 +408,6 @@ public void testTlsDisabled() throws Exception {
         PulsarClient pulsarClient = null;
 
         conf.setAuthenticationEnabled(false);
-        conf.setTlsEnabled(false);
         restartBroker();
 
         // Case 1: Access without TLS
@@ -447,7 +446,6 @@ public void testTlsEnabled() throws Exception {
         final String subName = "newSub";
 
         conf.setAuthenticationEnabled(false);
-        conf.setTlsEnabled(true);
         conf.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
         conf.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
         restartBroker();
@@ -525,7 +523,6 @@ public void testTlsAuthAllowInsecure() throws Exception {
 
         conf.setAuthenticationEnabled(true);
         conf.setAuthenticationProviders(providers);
-        conf.setTlsEnabled(true);
         conf.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
         conf.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
         conf.setTlsAllowInsecureConnection(true);
@@ -584,7 +581,6 @@ public void testTlsAuthDisallowInsecure() throws Exception {
 
         conf.setAuthenticationEnabled(true);
         conf.setAuthenticationProviders(providers);
-        conf.setTlsEnabled(true);
         conf.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
         conf.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
         conf.setTlsAllowInsecureConnection(false);
@@ -642,7 +638,6 @@ public void testTlsAuthUseTrustCert() throws Exception {
 
         conf.setAuthenticationEnabled(true);
         conf.setAuthenticationProviders(providers);
-        conf.setTlsEnabled(true);
         conf.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
         conf.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
         conf.setTlsAllowInsecureConnection(false);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
index b65b13fb44..9895e7bc52 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
@@ -1165,7 +1165,7 @@ public void testAtomicReplicationRemoval() throws Exception {
         ConcurrentOpenHashMap<String, Replicator> replicatorMap = topic.getReplicators();
 
         final URL brokerUrl = new URL(
-                "http://" + pulsar.getAdvertisedAddress() + ":" + pulsar.getConfiguration().getBrokerServicePort());
+                "http://" + pulsar.getAdvertisedAddress() + ":" + pulsar.getConfiguration().getBrokerServicePort().get());
         PulsarClient client = PulsarClient.builder().serviceUrl(brokerUrl.toString()).build();
         ManagedCursor cursor = mock(ManagedCursorImpl.class);
         doReturn(remoteCluster).when(cursor).getName();
@@ -1209,7 +1209,7 @@ public void testClosingReplicationProducerTwice() throws Exception {
         PersistentTopic topic = new PersistentTopic(globalTopicName, ledgerMock, brokerService);
 
         final URL brokerUrl = new URL(
-                "http://" + pulsar.getAdvertisedAddress() + ":" + pulsar.getConfiguration().getBrokerServicePort());
+                "http://" + pulsar.getAdvertisedAddress() + ":" + pulsar.getConfiguration().getBrokerServicePort().get());
         PulsarClient client = spy(PulsarClient.builder().serviceUrl(brokerUrl.toString()).build());
         PulsarClientImpl clientImpl = (PulsarClientImpl) client;
         doReturn(new CompletableFuture<Producer>()).when(clientImpl)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTestBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTestBase.java
index 10754b919e..8dd10f4b94 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTestBase.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTestBase.java
@@ -125,7 +125,6 @@ void setup() throws Exception {
                 inSec(getBrokerServicePurgeInactiveFrequency(), TimeUnit.SECONDS));
         config1.setBrokerServicePort(PortManager.nextFreePort());
         config1.setBrokerServicePortTls(PortManager.nextFreePort());
-        config1.setTlsEnabled(true);
         config1.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
         config1.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
         config1.setTlsTrustCertsFilePath(TLS_SERVER_CERT_FILE_PATH);
@@ -159,7 +158,6 @@ void setup() throws Exception {
                 inSec(getBrokerServicePurgeInactiveFrequency(), TimeUnit.SECONDS));
         config2.setBrokerServicePort(PortManager.nextFreePort());
         config2.setBrokerServicePortTls(PortManager.nextFreePort());
-        config2.setTlsEnabled(true);
         config2.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
         config2.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
         config2.setTlsTrustCertsFilePath(TLS_SERVER_CERT_FILE_PATH);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/v1/V1_ReplicatorTestBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/v1/V1_ReplicatorTestBase.java
index cdc7c06007..00b69a875a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/v1/V1_ReplicatorTestBase.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/v1/V1_ReplicatorTestBase.java
@@ -124,7 +124,6 @@ void setup() throws Exception {
                 inSec(getBrokerServicePurgeInactiveFrequency(), TimeUnit.SECONDS));
         config1.setBrokerServicePort(PortManager.nextFreePort());
         config1.setBrokerServicePortTls(PortManager.nextFreePort());
-        config1.setTlsEnabled(true);
         config1.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
         config1.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
         config1.setTlsTrustCertsFilePath(TLS_SERVER_CERT_FILE_PATH);
@@ -158,7 +157,6 @@ void setup() throws Exception {
                 inSec(getBrokerServicePurgeInactiveFrequency(), TimeUnit.SECONDS));
         config2.setBrokerServicePort(PortManager.nextFreePort());
         config2.setBrokerServicePortTls(PortManager.nextFreePort());
-        config2.setTlsEnabled(true);
         config2.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
         config2.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
         config2.setTlsTrustCertsFilePath(TLS_SERVER_CERT_FILE_PATH);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/WebServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/WebServiceTest.java
index 30e455412d..38cf4cc8aa 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/WebServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/WebServiceTest.java
@@ -243,13 +243,14 @@ private void setupEnv(boolean enableFilter, String minApiVersion, boolean allowU
         ServiceConfiguration config = new ServiceConfiguration();
         config.setAdvertisedAddress("localhost");
         config.setWebServicePort(BROKER_WEBSERVICE_PORT);
-        config.setWebServicePortTls(BROKER_WEBSERVICE_PORT_TLS);
+        if (enableTls) {
+            config.setWebServicePortTls(BROKER_WEBSERVICE_PORT_TLS);
+        }
         config.setClientLibraryVersionCheckEnabled(enableFilter);
         config.setAuthenticationEnabled(enableAuth);
         config.setAuthenticationProviders(providers);
         config.setAuthorizationEnabled(false);
         config.setSuperUserRoles(roles);
-        config.setTlsEnabled(enableTls);
         config.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
         config.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
         config.setTlsAllowInsecureConnection(allowInsecure);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthenticatedProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthenticatedProducerConsumerTest.java
index d2f6a8d057..4923894c66 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthenticatedProducerConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthenticatedProducerConsumerTest.java
@@ -71,7 +71,6 @@ protected void setup() throws Exception {
         conf.setAuthenticationEnabled(true);
         conf.setAuthorizationEnabled(true);
 
-        conf.setTlsEnabled(true);
         conf.setTlsTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH);
         conf.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
         conf.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthenticationTlsHostnameVerificationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthenticationTlsHostnameVerificationTest.java
index fef021e4c7..dbd893a9e1 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthenticationTlsHostnameVerificationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthenticationTlsHostnameVerificationTest.java
@@ -72,7 +72,6 @@ protected void setup() throws Exception {
         conf.setAuthenticationEnabled(true);
         conf.setAuthorizationEnabled(true);
 
-        conf.setTlsEnabled(true);
         conf.setTlsAllowInsecureConnection(true);
 
         Set<String> superUserRoles = new HashSet<>();
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 7cf4843824..54b10ba7c8 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
@@ -169,7 +169,7 @@ public void testMultipleBrokerLookup() throws Exception {
 
         /**** started broker-2 ****/
 
-        URI brokerServiceUrl = new URI("pulsar://localhost:" + conf2.getBrokerServicePort());
+        URI brokerServiceUrl = new URI("pulsar://localhost:" + conf2.getBrokerServicePort().get());
         PulsarClient pulsarClient2 = PulsarClient.builder().serviceUrl(brokerServiceUrl.toString()).build();
 
         // load namespace-bundle by calling Broker2
@@ -228,7 +228,7 @@ public void testMultipleBrokerDifferentClusterLookup() throws Exception {
         conf2.setAdvertisedAddress("localhost");
         conf2.setClusterName(newCluster); // Broker2 serves newCluster
         conf2.setZookeeperServers("localhost:2181");
-        String broker2ServiceUrl = "pulsar://localhost:" + conf2.getBrokerServicePort();
+        String broker2ServiceUrl = "pulsar://localhost:" + conf2.getBrokerServicePort().get();
 
         admin.clusters().createCluster(newCluster,
                 new ClusterData("http://127.0.0.1:" + BROKER_WEBSERVICE_PORT, null, broker2ServiceUrl, null));
@@ -394,7 +394,6 @@ public void testWebserviceServiceTls() throws Exception {
         conf2.setWebServicePortTls(PortManager.nextFreePort());
         conf2.setAdvertisedAddress("localhost");
         conf2.setTlsAllowInsecureConnection(true);
-        conf2.setTlsEnabled(true);
         conf2.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
         conf2.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
         conf2.setClusterName(conf.getClusterName());
@@ -403,7 +402,6 @@ public void testWebserviceServiceTls() throws Exception {
 
         // restart broker1 with tls enabled
         conf.setTlsAllowInsecureConnection(true);
-        conf.setTlsEnabled(true);
         conf.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
         conf.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
         stopBroker();
@@ -429,7 +427,7 @@ public void testWebserviceServiceTls() throws Exception {
 
         /**** started broker-2 ****/
 
-        URI brokerServiceUrl = new URI("pulsar://localhost:" + conf2.getBrokerServicePort());
+        URI brokerServiceUrl = new URI("pulsar://localhost:" + conf2.getBrokerServicePort().get());
         PulsarClient pulsarClient2 = PulsarClient.builder().serviceUrl(brokerServiceUrl.toString()).build();
 
         final String lookupResourceUrl = "/lookup/v2/topic/persistent/my-property/my-ns/my-topic1";
@@ -455,11 +453,11 @@ public void testWebserviceServiceTls() throws Exception {
         con.connect();
         log.info("connected url: {} ", con.getURL());
         // assert connect-url: broker2-https
-        assertEquals(con.getURL().getPort(), conf2.getWebServicePortTls());
+        assertEquals(new Integer(con.getURL().getPort()), conf2.getWebServicePortTls().get());
         InputStream is = con.getInputStream();
         // assert redirect-url: broker1-https only
         log.info("redirected url: {}", con.getURL());
-        assertEquals(con.getURL().getPort(), conf.getWebServicePortTls());
+        assertEquals(new Integer(con.getURL().getPort()), conf.getWebServicePortTls().get());
         is.close();
 
         pulsarClient2.close();
@@ -532,7 +530,6 @@ public void testDiscoveryLookupTls() throws Exception {
 
         // (1) restart broker1 with tls enabled
         conf.setTlsAllowInsecureConnection(true);
-        conf.setTlsEnabled(true);
         conf.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
         conf.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
         stopBroker();
@@ -542,7 +539,6 @@ public void testDiscoveryLookupTls() throws Exception {
         ServiceConfig config = new ServiceConfig();
         config.setServicePort(nextFreePort());
         config.setServicePortTls(nextFreePort());
-        config.setTlsEnabled(true);
         config.setBindOnLocalhost(true);
         config.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
         config.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
@@ -836,7 +832,7 @@ public void testSplitUnloadLookupTest() throws Exception {
         doReturn(Optional.of(resourceUnit)).when(loadManager1).getLeastLoaded(any(ServiceUnitId.class));
         loadManagerField.set(pulsar.getNamespaceService(), new AtomicReference<>(loadManager1));
 
-        URI broker2ServiceUrl = new URI("pulsar://localhost:" + conf2.getBrokerServicePort());
+        URI broker2ServiceUrl = new URI("pulsar://localhost:" + conf2.getBrokerServicePort().get());
         PulsarClient pulsarClient2 = PulsarClient.builder().serviceUrl(broker2ServiceUrl.toString()).build();
 
         // (3) Broker-2 receives topic-1 request, creates local-policies and sets the watch
@@ -946,7 +942,7 @@ public void testModularLoadManagerSplitBundle() throws Exception {
             doReturn(res).when(loadManager1).getLeastLoaded(any(ServiceUnitId.class));
             loadManagerField.set(pulsar.getNamespaceService(), new AtomicReference<>(loadManager1));
 
-            URI broker2ServiceUrl = new URI("pulsar://localhost:" + conf2.getBrokerServicePort());
+            URI broker2ServiceUrl = new URI("pulsar://localhost:" + conf2.getBrokerServicePort().get());
             PulsarClient pulsarClient2 = PulsarClient.builder().serviceUrl(broker2ServiceUrl.toString()).build();
 
             // (3) Broker-2 receives topic-1 request, creates local-policies and sets the watch
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ServiceUrlProviderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ServiceUrlProviderTest.java
index 4b73b9ef6d..73a51adba9 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ServiceUrlProviderTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ServiceUrlProviderTest.java
@@ -102,6 +102,8 @@ public void testCreateClientWithAutoChangedServiceUrlProvider() throws Exception
         PulsarService pulsarService1 = pulsar;
         conf.setBrokerServicePort(PortManager.nextFreePort());
         conf.setWebServicePort(PortManager.nextFreePort());
+        conf.setBrokerServicePortTls(PortManager.nextFreePort());
+        conf.setWebServicePortTls(PortManager.nextFreePort());
         startBroker();
         PulsarService pulsarService2 = pulsar;
         System.out.println("Pulsar1=" + pulsarService1.getBrokerServiceUrl() + ", Pulsar2=" + pulsarService2.getBrokerServiceUrl());
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsProducerConsumerBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsProducerConsumerBase.java
index 8e40011575..5d966cb6bb 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsProducerConsumerBase.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsProducerConsumerBase.java
@@ -58,7 +58,6 @@ protected void cleanup() throws Exception {
     }
 
     protected void internalSetUpForBroker() throws Exception {
-        conf.setTlsEnabled(true);
         conf.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
         conf.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
         conf.setTlsTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/common/naming/ServiceConfigurationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/common/naming/ServiceConfigurationTest.java
index 4bb4fb9945..fd51c06702 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/common/naming/ServiceConfigurationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/common/naming/ServiceConfigurationTest.java
@@ -56,7 +56,8 @@ public void testInit() throws Exception {
         InputStream newStream = updateProp(zookeeperServer, String.valueOf(brokerServicePort), "ns1,ns2");
         final ServiceConfiguration config = PulsarConfigurationLoader.create(newStream, ServiceConfiguration.class);
         assertTrue(isNotBlank(config.getZookeeperServers()));
-        assertTrue(config.getBrokerServicePort() == brokerServicePort);
+        assertTrue(config.getBrokerServicePort().isPresent()
+                && config.getBrokerServicePort().get().equals(brokerServicePort));
         assertEquals(config.getBootstrapNamespaces().get(1), "ns2");
     }
 
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarWorkerAssignmentTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarWorkerAssignmentTest.java
index 5210491eed..5cdf5a446a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarWorkerAssignmentTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarWorkerAssignmentTest.java
@@ -148,8 +148,8 @@ private WorkerService createPulsarFunctionWorker(ServiceConfiguration config) {
                 org.apache.pulsar.functions.worker.scheduler.RoundRobinScheduler.class.getName());
         workerConfig.setThreadContainerFactory(new WorkerConfig.ThreadContainerFactory().setThreadGroupName("use"));
         // worker talks to local broker
-        workerConfig.setPulsarServiceUrl("pulsar://127.0.0.1:" + config.getBrokerServicePort());
-        workerConfig.setPulsarWebServiceUrl("http://127.0.0.1:" + config.getWebServicePort());
+        workerConfig.setPulsarServiceUrl("pulsar://127.0.0.1:" + config.getBrokerServicePort().get());
+        workerConfig.setPulsarWebServiceUrl("http://127.0.0.1:" + config.getWebServicePort().get());
         workerConfig.setFailureCheckFreqMs(100);
         workerConfig.setNumFunctionPackageReplicas(1);
         workerConfig.setClusterCoordinationTopicName("coordinate");
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionAdminTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionAdminTest.java
index a2a5b283cd..b9c21628fd 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionAdminTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionAdminTest.java
@@ -117,7 +117,6 @@ void setup(Method method) throws Exception {
         providers.add(AuthenticationProviderTls.class.getName());
         config.setAuthenticationEnabled(true);
         config.setAuthenticationProviders(providers);
-        config.setTlsEnabled(true);
         config.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
         config.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
         config.setTlsAllowInsecureConnection(true);
@@ -180,8 +179,8 @@ private WorkerService createPulsarFunctionWorker(ServiceConfiguration config) {
                 org.apache.pulsar.functions.worker.scheduler.RoundRobinScheduler.class.getName());
         workerConfig.setThreadContainerFactory(new WorkerConfig.ThreadContainerFactory().setThreadGroupName("use"));
         // worker talks to local broker
-        workerConfig.setPulsarServiceUrl("pulsar://127.0.0.1:" + config.getBrokerServicePortTls());
-        workerConfig.setPulsarWebServiceUrl("https://127.0.0.1:" + config.getWebServicePortTls());
+        workerConfig.setPulsarServiceUrl("pulsar://127.0.0.1:" + config.getBrokerServicePortTls().get());
+        workerConfig.setPulsarWebServiceUrl("https://127.0.0.1:" + config.getWebServicePortTls().get());
         workerConfig.setFailureCheckFreqMs(100);
         workerConfig.setNumFunctionPackageReplicas(1);
         workerConfig.setClusterCoordinationTopicName("coordinate");
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionE2ETest.java b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionE2ETest.java
index 1980bced1b..7841453c20 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionE2ETest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionE2ETest.java
@@ -149,7 +149,6 @@ void setup(Method method) throws Exception {
         providers.add(AuthenticationProviderTls.class.getName());
         config.setAuthenticationEnabled(true);
         config.setAuthenticationProviders(providers);
-        config.setTlsEnabled(true);
         config.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
         config.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
         config.setTlsAllowInsecureConnection(true);
@@ -212,8 +211,8 @@ private WorkerService createPulsarFunctionWorker(ServiceConfiguration config) {
                 org.apache.pulsar.functions.worker.scheduler.RoundRobinScheduler.class.getName());
         workerConfig.setThreadContainerFactory(new WorkerConfig.ThreadContainerFactory().setThreadGroupName("use"));
         // worker talks to local broker
-        workerConfig.setPulsarServiceUrl("pulsar://127.0.0.1:" + config.getBrokerServicePortTls());
-        workerConfig.setPulsarWebServiceUrl("https://127.0.0.1:" + config.getWebServicePortTls());
+        workerConfig.setPulsarServiceUrl("pulsar://127.0.0.1:" + config.getBrokerServicePortTls().get());
+        workerConfig.setPulsarWebServiceUrl("https://127.0.0.1:" + config.getWebServicePortTls().get());
         workerConfig.setFailureCheckFreqMs(100);
         workerConfig.setNumFunctionPackageReplicas(1);
         workerConfig.setClusterCoordinationTopicName("coordinate");
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionTlsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionTlsTest.java
index fb081e5bf7..a37786ff07 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionTlsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionTlsTest.java
@@ -106,7 +106,6 @@ void setup(Method method) throws Exception {
         providers.add(AuthenticationProviderTls.class.getName());
         config.setAuthenticationEnabled(true);
         config.setAuthenticationProviders(providers);
-        config.setTlsEnabled(true);
         config.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
         config.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
         config.setTlsAllowInsecureConnection(true);
@@ -166,8 +165,8 @@ private WorkerService createPulsarFunctionWorker(ServiceConfiguration config) {
                 org.apache.pulsar.functions.worker.scheduler.RoundRobinScheduler.class.getName());
         workerConfig.setThreadContainerFactory(new WorkerConfig.ThreadContainerFactory().setThreadGroupName("use"));
         // worker talks to local broker
-        workerConfig.setPulsarServiceUrl("pulsar://127.0.0.1:" + config.getBrokerServicePortTls());
-        workerConfig.setPulsarWebServiceUrl("https://127.0.0.1:" + config.getWebServicePortTls());
+        workerConfig.setPulsarServiceUrl("pulsar://127.0.0.1:" + config.getBrokerServicePort().get());
+        workerConfig.setPulsarWebServiceUrl("https://127.0.0.1:" + config.getWebServicePort().get());
         workerConfig.setFailureCheckFreqMs(100);
         workerConfig.setNumFunctionPackageReplicas(1);
         workerConfig.setClusterCoordinationTopicName("coordinate");
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java
index 9ddf955484..6f08cc8595 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java
@@ -407,7 +407,7 @@ public void testProxyStats() throws Exception {
 
             Client client = ClientBuilder.newClient(new ClientConfig().register(LoggingFeature.class));
             final String baseUrl = pulsar.getWebServiceAddress()
-                    .replace(Integer.toString(pulsar.getConfiguration().getWebServicePort()), (Integer.toString(port)))
+                    .replace(Integer.toString(pulsar.getConfiguration().getWebServicePort().get()), (Integer.toString(port)))
                     + "/admin/v2/proxy-stats/";
 
             // verify proxy metrics
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTlsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTlsTest.java
index 650e05e139..96f3c758e9 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTlsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTlsTest.java
@@ -67,7 +67,7 @@ public void setup() throws Exception {
         WebSocketProxyConfiguration config = new WebSocketProxyConfiguration();
         config.setWebServicePort(port);
         config.setWebServicePortTls(tlsPort);
-        config.setTlsEnabled(true);
+        config.setBrokerClientTlsEnabled(true);
         config.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
         config.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
         config.setTlsTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH);
diff --git a/pulsar-discovery-service/src/main/java/org/apache/pulsar/discovery/service/DiscoveryService.java b/pulsar-discovery-service/src/main/java/org/apache/pulsar/discovery/service/DiscoveryService.java
index c02baaa1ef..ee8531d3b0 100644
--- a/pulsar-discovery-service/src/main/java/org/apache/pulsar/discovery/service/DiscoveryService.java
+++ b/pulsar-discovery-service/src/main/java/org/apache/pulsar/discovery/service/DiscoveryService.java
@@ -18,6 +18,7 @@
  */
 package org.apache.pulsar.discovery.service;
 
+import com.google.common.base.Preconditions;
 import static com.google.common.base.Preconditions.checkNotNull;
 
 import java.io.Closeable;
@@ -104,15 +105,23 @@ public void startServer() throws Exception {
 
         bootstrap.childHandler(new ServiceChannelInitializer(this, config, false));
         // Bind and start to accept incoming connections.
-        bootstrap.bind(config.getServicePort()).sync();
-        LOG.info("Started Pulsar Discovery service on port {}", config.getServicePort());
-
-        if (config.isTlsEnabled()) {
+        
+        Preconditions.checkArgument(config.getServicePort().isPresent() || config.getServicePortTls().isPresent(), 
+                "Either ServicePort or ServicePortTls should be configured.");
+        
+        if (config.getServicePort().isPresent()) {
+            // Bind and start to accept incoming connections.
+            bootstrap.bind(config.getServicePort().get()).sync();
+            LOG.info("Started Pulsar Discovery service on port {}", config.getServicePort());
+        }
+        
+        if (config.getServicePortTls().isPresent()) {
             ServerBootstrap tlsBootstrap = bootstrap.clone();
             tlsBootstrap.childHandler(new ServiceChannelInitializer(this, config, true));
-            tlsBootstrap.bind(config.getServicePortTls()).sync();
-            LOG.info("Started Pulsar Discovery TLS service on port {}", config.getServicePortTls());
+            tlsBootstrap.bind(config.getServicePortTls().get()).sync();
+            LOG.info("Started Pulsar Discovery TLS service on port {}", config.getServicePortTls().get());
         }
+        
     }
 
     public ZooKeeperClientFactory getZooKeeperClientFactory() {
@@ -153,15 +162,20 @@ public String host() {
     }
 
     public String serviceUrl() {
-        return new StringBuilder("pulsar://").append(host()).append(":").append(config.getServicePort()).toString();
+        if (config.getServicePort().isPresent()) {
+            return new StringBuilder("pulsar://").append(host()).append(":").append(config.getServicePort().get())
+                    .toString();
+        } else {
+            return null;
+        }
     }
 
     public String serviceUrlTls() {
-        if (config.isTlsEnabled()) {
-            return new StringBuilder("pulsar+ssl://").append(host()).append(":").append(config.getServicePortTls())
+        if (config.getServicePortTls().isPresent()) {
+            return new StringBuilder("pulsar+ssl://").append(host()).append(":").append(config.getServicePortTls().get())
                     .toString();
         } else {
-            return "";
+            return null;
         }
     }
 
diff --git a/pulsar-discovery-service/src/main/java/org/apache/pulsar/discovery/service/server/ServerManager.java b/pulsar-discovery-service/src/main/java/org/apache/pulsar/discovery/service/server/ServerManager.java
index c87cd7da60..4cd32396dc 100644
--- a/pulsar-discovery-service/src/main/java/org/apache/pulsar/discovery/service/server/ServerManager.java
+++ b/pulsar-discovery-service/src/main/java/org/apache/pulsar/discovery/service/server/ServerManager.java
@@ -53,21 +53,21 @@
     private final Server server;
     private final ExecutorThreadPool webServiceExecutor;
     private final List<Handler> handlers = Lists.newArrayList();
-    protected final int externalServicePort;
 
     public ServerManager(ServiceConfig config) {
         this.webServiceExecutor = new ExecutorThreadPool();
         this.webServiceExecutor.setName("pulsar-external-web");
         this.server = new Server(webServiceExecutor);
-        this.externalServicePort = config.getWebServicePort();
 
         List<ServerConnector> connectors = Lists.newArrayList();
 
-        ServerConnector connector = new ServerConnector(server, 1, 1);
-        connector.setPort(externalServicePort);
-        connectors.add(connector);
+        if (config.getWebServicePort().isPresent()) {
+            ServerConnector connector = new ServerConnector(server, 1, 1);
+            connector.setPort(config.getWebServicePort().get());
+            connectors.add(connector);
+        }
 
-        if (config.isTlsEnabled()) {
+        if (config.getWebServicePortTls().isPresent()) {
             try {
                 SslContextFactory sslCtxFactory = SecurityUtility.createSslContextFactory(
                         config.isTlsAllowInsecureConnection(),
@@ -76,7 +76,7 @@ public ServerManager(ServiceConfig config) {
                         config.getTlsKeyFilePath(), 
                         config.getTlsRequireTrustedClientCertOnConnect());
                 ServerConnector tlsConnector = new ServerConnector(server, 1, 1, sslCtxFactory);
-                tlsConnector.setPort(config.getWebServicePortTls());
+                tlsConnector.setPort(config.getWebServicePortTls().get());
                 connectors.add(tlsConnector);
             } catch (GeneralSecurityException e) {
                 throw new RestException(e);
@@ -102,10 +102,6 @@ public void addServlet(String path, Class<? extends Servlet> servlet, Map<String
         handlers.add(context);
     }
 
-    public int getExternalServicePort() {
-        return externalServicePort;
-    }
-
     public void start() throws Exception {
         RequestLogHandler requestLogHandler = new RequestLogHandler();
         Slf4jRequestLog requestLog = new Slf4jRequestLog();
diff --git a/pulsar-discovery-service/src/main/java/org/apache/pulsar/discovery/service/server/ServiceConfig.java b/pulsar-discovery-service/src/main/java/org/apache/pulsar/discovery/service/server/ServiceConfig.java
index c024c19881..bc7900eb27 100644
--- a/pulsar-discovery-service/src/main/java/org/apache/pulsar/discovery/service/server/ServiceConfig.java
+++ b/pulsar-discovery-service/src/main/java/org/apache/pulsar/discovery/service/server/ServiceConfig.java
@@ -18,6 +18,7 @@
  */
 package org.apache.pulsar.discovery.service.server;
 
+import java.util.Optional;
 import java.util.Properties;
 import java.util.Set;
 
@@ -45,13 +46,13 @@
     private int zookeeperSessionTimeoutMs = 30_000;
 
     // Port to use to server binary-proto request
-    private int servicePort = 5000;
+    private Integer servicePort = 5000;
     // Port to use to server binary-proto-tls request
-    private int servicePortTls = 5001;
+    private Integer servicePortTls;
     // Port to use to server HTTP request
-    private int webServicePort = 8080;
+    private Integer webServicePort = 8080;
     // Port to use to server HTTPS request
-    private int webServicePortTls = 8443;
+    private Integer webServicePortTls;
     // Control whether to bind directly on localhost rather than on normal
     // hostname
     private boolean bindOnLocalhost = false;
@@ -75,7 +76,7 @@
     private String authorizationProvider = PulsarAuthorizationProvider.class.getName();
 
     /***** --- TLS --- ****/
-    // Enable TLS
+    @Deprecated
     private boolean tlsEnabled = false;
     // Path for the TLS certificate file
     private String tlsCertificateFilePath;
@@ -131,42 +132,44 @@ public void setZookeeperSessionTimeoutMs(int zookeeperSessionTimeoutMs) {
         this.zookeeperSessionTimeoutMs = zookeeperSessionTimeoutMs;
     }
 
-    public int getServicePort() {
-        return servicePort;
+    public Optional<Integer> getServicePort() {
+        return Optional.ofNullable(servicePort);
     }
 
     public void setServicePort(int servicePort) {
         this.servicePort = servicePort;
     }
 
-    public int getServicePortTls() {
-        return servicePortTls;
+    public Optional<Integer> getServicePortTls() {
+        return Optional.ofNullable(servicePortTls);
     }
 
     public void setServicePortTls(int servicePortTls) {
         this.servicePortTls = servicePortTls;
     }
 
-    public int getWebServicePort() {
-        return webServicePort;
+    public Optional<Integer> getWebServicePort() {
+        return Optional.ofNullable(webServicePort);
     }
 
     public void setWebServicePort(int webServicePort) {
         this.webServicePort = webServicePort;
     }
 
-    public int getWebServicePortTls() {
-        return webServicePortTls;
+    public Optional<Integer> getWebServicePortTls() {
+        return Optional.ofNullable(webServicePortTls);
     }
 
     public void setWebServicePortTls(int webServicePortTls) {
         this.webServicePortTls = webServicePortTls;
     }
 
+    @Deprecated
     public boolean isTlsEnabled() {
-        return tlsEnabled;
+        return tlsEnabled || webServicePortTls != null || servicePortTls != null;
     }
 
+    @Deprecated
     public void setTlsEnabled(boolean tlsEnabled) {
         this.tlsEnabled = tlsEnabled;
     }
diff --git a/pulsar-discovery-service/src/test/java/org/apache/pulsar/discovery/service/BaseDiscoveryTestSetup.java b/pulsar-discovery-service/src/test/java/org/apache/pulsar/discovery/service/BaseDiscoveryTestSetup.java
index 540d3b826e..88e4a7fe0d 100644
--- a/pulsar-discovery-service/src/test/java/org/apache/pulsar/discovery/service/BaseDiscoveryTestSetup.java
+++ b/pulsar-discovery-service/src/test/java/org/apache/pulsar/discovery/service/BaseDiscoveryTestSetup.java
@@ -53,7 +53,6 @@ protected void setup() throws Exception {
         config.setServicePortTls(nextFreePort());
         config.setBindOnLocalhost(true);
 
-        config.setTlsEnabled(true);
         config.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
         config.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
 
diff --git a/pulsar-discovery-service/src/test/java/org/apache/pulsar/discovery/service/web/DiscoveryServiceWebTest.java b/pulsar-discovery-service/src/test/java/org/apache/pulsar/discovery/service/web/DiscoveryServiceWebTest.java
index 648f57415a..452daaae88 100644
--- a/pulsar-discovery-service/src/test/java/org/apache/pulsar/discovery/service/web/DiscoveryServiceWebTest.java
+++ b/pulsar-discovery-service/src/test/java/org/apache/pulsar/discovery/service/web/DiscoveryServiceWebTest.java
@@ -194,7 +194,6 @@ public void testTlsEnable() throws Exception {
         ServiceConfig config = new ServiceConfig();
         config.setWebServicePort(port);
         config.setWebServicePortTls(tlsPort);
-        config.setTlsEnabled(true);
         config.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
         config.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
         ServerManager server = new ServerManager(config);
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
index 0cc214cc87..d49722f871 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
@@ -55,8 +55,8 @@
 
     private String workerId;
     private String workerHostname;
-    private int workerPort;
-    private int workerPortTls;
+    private Integer workerPort;
+    private Integer workerPortTls;
     private String connectorsDirectory = "./connectors";
     private String functionMetadataTopicName;
     private String functionWebServiceUrl;
@@ -80,7 +80,7 @@
     // Frequency how often worker performs compaction on function-topics
     private long topicCompactionFrequencySec = 30 * 60; // 30 minutes
     /***** --- TLS --- ****/
-    // Enable TLS
+    @Deprecated
     private boolean tlsEnabled = false;
     // Path for the TLS certificate file
     private String tlsCertificateFilePath;
@@ -91,6 +91,7 @@
     // Accept untrusted TLS certificate from client
     private boolean tlsAllowInsecureConnection = false;
     private boolean tlsRequireTrustedClientCertOnConnect = false;
+    // TLS for Functions -> Broker
     private boolean useTls = false;
     private boolean tlsHostnameVerificationEnable = false;
     // Enforce authentication
@@ -104,6 +105,11 @@
     
     private Properties properties = new Properties();
 
+    public boolean getTlsEnabled() {
+    	return tlsEnabled || workerPortTls != null;
+    }
+    
+    
     @Data
     @Setter
     @Getter
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/WorkerServer.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/WorkerServer.java
index 2a3e04cf51..0e280377e2 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/WorkerServer.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/WorkerServer.java
@@ -115,7 +115,7 @@ private void init() {
         handlerCollection.setHandlers(new Handler[] { contexts, new DefaultHandler(), requestLogHandler });
         server.setHandler(handlerCollection);
 
-        if (this.workerConfig.isTlsEnabled()) {
+        if (this.workerConfig.getWorkerPortTls() != null) {
             try {
                 SslContextFactory sslCtxFactory = SecurityUtility.createSslContextFactory(
                         this.workerConfig.isTlsAllowInsecureConnection(), this.workerConfig.getTlsTrustCertsFilePath(),
diff --git a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/WorkerApiV2ResourceConfigTest.java b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/WorkerApiV2ResourceConfigTest.java
index 4c6d24912f..1d60aa41f0 100644
--- a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/WorkerApiV2ResourceConfigTest.java
+++ b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/WorkerApiV2ResourceConfigTest.java
@@ -45,7 +45,7 @@ public void testGetterSetter() {
         assertEquals("sample/standalone/functions", wc.getPulsarFunctionsNamespace());
         assertEquals(3, wc.getNumFunctionPackageReplicas());
         assertEquals(TEST_NAME + "-worker", wc.getWorkerId());
-        assertEquals(1234, wc.getWorkerPort());
+        assertEquals(new Integer(1234), wc.getWorkerPort());
     }
 
     @Test
@@ -57,7 +57,7 @@ public void testLoadWorkerConfig() throws Exception {
         assertEquals("test-function-metadata-topic", wc.getFunctionMetadataTopicName());
         assertEquals(3, wc.getNumFunctionPackageReplicas());
         assertEquals("test-worker", wc.getWorkerId());
-        assertEquals(7654, wc.getWorkerPort());
+        assertEquals(new Integer(7654), wc.getWorkerPort());
     }
 
 }
diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConfiguration.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConfiguration.java
index 45eba27c9d..717019b8df 100644
--- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConfiguration.java
+++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConfiguration.java
@@ -23,6 +23,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Properties;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -129,23 +130,23 @@
         category = CATEGORY_SERVER,
         doc = "The port for serving binary protobuf request"
     )
-    private int servicePort = 6650;
+    private Integer servicePort = 6650;
     @FieldContext(
         category = CATEGORY_SERVER,
         doc = "The port for serving tls secured binary protobuf request"
     )
-    private int servicePortTls = 6651;
+    private Integer servicePortTls;
 
     @FieldContext(
         category = CATEGORY_SERVER,
         doc = "The port for serving http requests"
     )
-    private int webServicePort = 8080;
+    private Integer webServicePort = 8080;
     @FieldContext(
         category = CATEGORY_SERVER,
         doc = "The port for serving https requests"
     )
-    private int webServicePortTls = 8443;
+    private Integer webServicePortTls;
 
     @FieldContext(
         category = CATEGORY_SERVER,
@@ -225,10 +226,8 @@
     private boolean tlsEnabledWithBroker = false;
 
     /***** --- TLS --- ****/
-    @FieldContext(
-        category = CATEGORY_TLS,
-        doc = "Whether TLS is enabled for the proxy"
-    )
+
+    @Deprecated 
     private boolean tlsEnabledInProxy = false;
 
     @FieldContext(
@@ -332,6 +331,22 @@ public Properties getProperties() {
         return properties;
     }
 
+    public Optional<Integer> getServicePort() {
+        return Optional.ofNullable(servicePort);
+    }
+
+    public Optional<Integer> getServicePortTls() {
+        return Optional.ofNullable(servicePortTls);
+    }
+    
+    public Optional<Integer> getWebServicePort() {
+        return Optional.ofNullable(webServicePort);
+    }
+    
+    public Optional<Integer> getWebServicePortTls() {
+        return Optional.ofNullable(webServicePortTls);
+    }
+    
     public void setProperties(Properties properties) {
         this.properties = properties;
 
@@ -387,4 +402,4 @@ public String toString() {
             return String.format("HttpReverseProxyConfig(%s, path=%s, proxyTo=%s)", name, path, proxyTo);
         }
     }
-}
+}
\ No newline at end of file
diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyService.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyService.java
index eaf5f04fe7..317daa39f3 100644
--- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyService.java
+++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyService.java
@@ -104,8 +104,17 @@ public ProxyService(ProxyConfiguration proxyConfig,
         } catch (UnknownHostException e) {
             throw new RuntimeException(e);
         }
-        this.serviceUrl = String.format("pulsar://%s:%d/", hostname, proxyConfig.getServicePort());
-        this.serviceUrlTls = String.format("pulsar://%s:%d/", hostname, proxyConfig.getServicePortTls());
+        if (proxyConfig.getServicePort().isPresent()) {
+            this.serviceUrl = String.format("pulsar://%s:%d/", hostname, proxyConfig.getServicePort().get());
+        } else {
+            this.serviceUrl = null;
+        }
+        
+        if (proxyConfig.getServicePortTls().isPresent()) {
+            this.serviceUrlTls = String.format("pulsar://%s:%d/", hostname, proxyConfig.getServicePortTls().get());
+        } else {
+            this.serviceUrlTls = null;
+        }
 
         this.acceptorGroup = EventLoopUtil.newEventLoopGroup(1, acceptorThreadFactory);
         this.workerGroup = EventLoopUtil.newEventLoopGroup(numThreads, workersThreadFactory);
@@ -132,18 +141,21 @@ public void start() throws Exception {
 
         bootstrap.childHandler(new ServiceChannelInitializer(this, proxyConfig, false));
         // Bind and start to accept incoming connections.
-        try {
-            bootstrap.bind(proxyConfig.getServicePort()).sync();
-        } catch (Exception e) {
-            throw new IOException("Failed to bind Pulsar Proxy on port " + proxyConfig.getServicePort(), e);
+        if (proxyConfig.getServicePort().isPresent()) {
+            try {
+                bootstrap.bind(proxyConfig.getServicePort().get()).sync();
+                LOG.info("Started Pulsar Proxy at {}", serviceUrl);
+            } catch (Exception e) {
+                throw new IOException("Failed to bind Pulsar Proxy on port " + proxyConfig.getServicePort().get(), e);
+            }
         }
         LOG.info("Started Pulsar Proxy at {}", serviceUrl);
 
-        if (proxyConfig.isTlsEnabledInProxy()) {
+        if (proxyConfig.getServicePortTls().isPresent()) {
             ServerBootstrap tlsBootstrap = bootstrap.clone();
             tlsBootstrap.childHandler(new ServiceChannelInitializer(this, proxyConfig, true));
-            tlsBootstrap.bind(proxyConfig.getServicePortTls()).sync();
-            LOG.info("Started Pulsar TLS Proxy on port {}", proxyConfig.getServicePortTls());
+            tlsBootstrap.bind(proxyConfig.getServicePortTls().get()).sync();
+            LOG.info("Started Pulsar TLS Proxy on port {}", proxyConfig.getServicePortTls().get());
         }
     }
 
diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/WebServer.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/WebServer.java
index 07d75744b7..20f6b56a90 100644
--- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/WebServer.java
+++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/WebServer.java
@@ -82,7 +82,6 @@ public WebServer(ProxyConfiguration config, AuthenticationService authentication
         this.webServiceExecutor = new ExecutorThreadPool();
         this.webServiceExecutor.setName("pulsar-external-web");
         this.server = new Server(webServiceExecutor);
-        this.externalServicePort = config.getWebServicePort();
         this.authenticationService = authenticationService;
         this.config = config;
 
@@ -91,11 +90,13 @@ public WebServer(ProxyConfiguration config, AuthenticationService authentication
         HttpConfiguration http_config = new HttpConfiguration();
         http_config.setOutputBufferSize(config.getHttpOutputBufferSize());
 
-        ServerConnector connector = new ServerConnector(server, 1, 1, new HttpConnectionFactory(http_config));
-        connector.setPort(externalServicePort);
-        connectors.add(connector);
-
-        if (config.isTlsEnabledInProxy()) {
+        if (config.getWebServicePort().isPresent()) {
+            this.externalServicePort = config.getWebServicePort().get();
+            ServerConnector connector = new ServerConnector(server, 1, 1, new HttpConnectionFactory(http_config));
+            connector.setPort(externalServicePort);
+            connectors.add(connector);
+        }
+        if (config.getWebServicePortTls().isPresent()) {
             try {
                 SslContextFactory sslCtxFactory = SecurityUtility.createSslContextFactory(
                         config.isTlsAllowInsecureConnection(),
@@ -104,7 +105,7 @@ public WebServer(ProxyConfiguration config, AuthenticationService authentication
                         config.getTlsKeyFilePath(),
                         config.isTlsRequireTrustedClientCertOnConnect());
                 ServerConnector tlsConnector = new ServerConnector(server, 1, 1, sslCtxFactory);
-                tlsConnector.setPort(config.getWebServicePortTls());
+                tlsConnector.setPort(config.getWebServicePortTls().get());
                 connectors.add(tlsConnector);
             } catch (GeneralSecurityException e) {
                 throw new RuntimeException(e);
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/AuthedAdminProxyHandlerTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/AuthedAdminProxyHandlerTest.java
index 7c638f4d83..096ff051b5 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/AuthedAdminProxyHandlerTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/AuthedAdminProxyHandlerTest.java
@@ -62,7 +62,6 @@ protected void setup() throws Exception {
         // enable tls and auth&auth at broker
         conf.setAuthenticationEnabled(true);
         conf.setAuthorizationEnabled(true);
-        conf.setTlsEnabled(true);
         conf.setTlsTrustCertsFilePath(getTlsFile("ca.cert"));
         conf.setTlsCertificateFilePath(getTlsFile("broker.cert"));
         conf.setTlsKeyFilePath(getTlsFile("broker.key-pk8"));
@@ -80,7 +79,6 @@ protected void setup() throws Exception {
         proxyConfig.setServicePortTls(PortManager.nextFreePort());
         proxyConfig.setWebServicePort(PortManager.nextFreePort());
         proxyConfig.setWebServicePortTls(PortManager.nextFreePort());
-        proxyConfig.setTlsEnabledInProxy(true);
         proxyConfig.setTlsEnabledWithBroker(true);
 
         // enable tls and auth&auth at proxy
@@ -130,7 +128,7 @@ PulsarAdmin getDirectToBrokerAdminClient(String user) throws Exception {
 
     PulsarAdmin getAdminClient(String user) throws Exception {
         return PulsarAdmin.builder()
-            .serviceHttpUrl("https://localhost:" + proxyConfig.getWebServicePortTls())
+            .serviceHttpUrl("https://localhost:" + proxyConfig.getWebServicePortTls().get())
             .tlsTrustCertsFilePath(getTlsFile("ca.cert"))
             .allowTlsInsecureConnection(false)
             .authentication(AuthenticationTls.class.getName(),
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyAuthenticatedProducerConsumerTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyAuthenticatedProducerConsumerTest.java
index c31773fcb0..50cd920e6e 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyAuthenticatedProducerConsumerTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyAuthenticatedProducerConsumerTest.java
@@ -74,7 +74,6 @@ protected void setup() throws Exception {
         conf.setAuthenticationEnabled(true);
         conf.setAuthorizationEnabled(true);
 
-        conf.setTlsEnabled(true);
         conf.setTlsTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH);
         conf.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
         conf.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
@@ -105,7 +104,6 @@ protected void setup() throws Exception {
         proxyConfig.setServicePortTls(PortManager.nextFreePort());
         proxyConfig.setWebServicePort(PortManager.nextFreePort());
         proxyConfig.setWebServicePortTls(PortManager.nextFreePort());
-        proxyConfig.setTlsEnabledInProxy(true);
         proxyConfig.setTlsEnabledWithBroker(true);
 
         // enable tls and auth&auth at proxy
@@ -154,7 +152,7 @@ protected void cleanup() throws Exception {
     public void testTlsSyncProducerAndConsumer() throws Exception {
         log.info("-- Starting {} test --", methodName);
 
-        final String proxyServiceUrl = "pulsar://localhost:" + proxyConfig.getServicePortTls();
+        final String proxyServiceUrl = "pulsar://localhost:" + proxyConfig.getServicePortTls().get();
         Map<String, String> authParams = Maps.newHashMap();
         authParams.put("tlsCertFile", TLS_CLIENT_CERT_FILE_PATH);
         authParams.put("tlsKeyFile", TLS_CLIENT_KEY_FILE_PATH);
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyAuthenticationTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyAuthenticationTest.java
index cd3862877e..246078a21c 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyAuthenticationTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyAuthenticationTest.java
@@ -174,7 +174,6 @@ protected void setup() throws Exception {
 		servicePort = PortManager.nextFreePort();
 		conf.setAuthenticationEnabled(true);
 		conf.setAuthorizationEnabled(true);
-		conf.setTlsEnabled(false);
 		conf.setBrokerClientAuthenticationPlugin(BasicAuthentication.class.getName());
 		// Expires after an hour
 		conf.setBrokerClientAuthenticationParameters(
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyConnectionThrottlingTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyConnectionThrottlingTest.java
index 20ebeb1187..4813615a4b 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyConnectionThrottlingTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyConnectionThrottlingTest.java
@@ -70,12 +70,12 @@ protected void cleanup() throws Exception {
     @Test
     public void testInboundConnection() throws Exception {
         LOG.info("Creating producer 1");
-        PulsarClient client1 = PulsarClient.builder().serviceUrl("pulsar://localhost:" + proxyConfig.getServicePort())
+        PulsarClient client1 = PulsarClient.builder().serviceUrl("pulsar://localhost:" + proxyConfig.getServicePort().get())
                 .build();
         Producer<byte[]> producer1 = client1.newProducer(Schema.BYTES).topic("persistent://sample/test/local/producer-topic-1").create();
 
         LOG.info("Creating producer 2");
-        PulsarClient client2 = PulsarClient.builder().serviceUrl("pulsar://localhost:" + proxyConfig.getServicePort())
+        PulsarClient client2 = PulsarClient.builder().serviceUrl("pulsar://localhost:" + proxyConfig.getServicePort().get())
                 .build();
         Producer<byte[]> producer2;
         Assert.assertEquals(ProxyService.rejectedConnections.get(), 0.0d);
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyForwardAuthDataTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyForwardAuthDataTest.java
index 42915ec2ce..4997de0165 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyForwardAuthDataTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyForwardAuthDataTest.java
@@ -54,7 +54,6 @@ protected void setup() throws Exception {
         servicePort = PortManager.nextFreePort();
         conf.setAuthenticationEnabled(true);
         conf.setAuthorizationEnabled(true);
-        conf.setTlsEnabled(false);
         conf.setBrokerClientAuthenticationPlugin(BasicAuthentication.class.getName());
         conf.setBrokerClientAuthenticationParameters("authParam:broker");
         conf.setAuthenticateOriginalAuthData(true);
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyLookupThrottlingTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyLookupThrottlingTest.java
index 35dfb94374..9c8ea850c3 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyLookupThrottlingTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyLookupThrottlingTest.java
@@ -70,7 +70,7 @@ protected void cleanup() throws Exception {
 
     @Test
     public void testLookup() throws Exception {
-        PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://localhost:" + proxyConfig.getServicePort())
+        PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://localhost:" + proxyConfig.getServicePort().get())
                 .connectionsPerBroker(5).ioThreads(5).build();
         assertTrue(proxyService.getLookupRequestSemaphore().tryAcquire());
         assertTrue(proxyService.getLookupRequestSemaphore().tryAcquire());
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyRolesEnforcementTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyRolesEnforcementTest.java
index 69f4f5a583..7d93ca6d0d 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyRolesEnforcementTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyRolesEnforcementTest.java
@@ -152,7 +152,6 @@ protected void setup() throws Exception {
         servicePort = PortManager.nextFreePort();
         conf.setAuthenticationEnabled(true);
         conf.setAuthorizationEnabled(true);
-        conf.setTlsEnabled(false);
         conf.setBrokerClientAuthenticationPlugin(BasicAuthentication.class.getName());
         conf.setBrokerClientAuthenticationParameters("authParam:broker");
 
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyTest.java
index 09fdbdb159..be47d9ea09 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyTest.java
@@ -91,7 +91,7 @@ protected void cleanup() throws Exception {
 
     @Test
     public void testProducer() throws Exception {
-        PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://localhost:" + proxyConfig.getServicePort())
+        PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://localhost:" + proxyConfig.getServicePort().get())
                 .build();
         Producer<byte[]> producer = client.newProducer(Schema.BYTES).topic("persistent://sample/test/local/producer-topic")
                 .create();
@@ -105,7 +105,7 @@ public void testProducer() throws Exception {
 
     @Test
     public void testProducerConsumer() throws Exception {
-        PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://localhost:" + proxyConfig.getServicePort())
+        PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://localhost:" + proxyConfig.getServicePort().get())
                 .build();
         Producer<byte[]> producer = client.newProducer(Schema.BYTES)
             .topic("persistent://sample/test/local/producer-consumer-topic")
@@ -137,7 +137,7 @@ public void testProducerConsumer() throws Exception {
     @Test
     public void testPartitions() throws Exception {
         admin.tenants().createTenant("sample", new TenantInfo());
-        PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://localhost:" + proxyConfig.getServicePort())
+        PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://localhost:" + proxyConfig.getServicePort().get())
                 .build();
         admin.topics().createPartitionedTopic("persistent://sample/test/local/partitioned-topic", 2);
 
@@ -164,7 +164,7 @@ public void testPartitions() throws Exception {
 
     @Test
     public void testRegexSubscription() throws Exception {
-        PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://localhost:" + proxyConfig.getServicePort())
+        PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://localhost:" + proxyConfig.getServicePort().get())
             .connectionsPerBroker(5).ioThreads(5).build();
 
         // create two topics by subscribing to a topic and closing it
@@ -207,7 +207,7 @@ public void testRegexSubscription() throws Exception {
 
     @Test
     private void testProtocolVersionAdvertisement() throws Exception {
-        final String url = "pulsar://localhost:" + proxyConfig.getServicePort();
+        final String url = "pulsar://localhost:" + proxyConfig.getServicePort().get();
         final String topic = "persistent://sample/test/local/protocol-version-advertisement";
         final String sub = "my-sub";
 
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyTlsTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyTlsTest.java
index db5070fd32..2bf544d92c 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyTlsTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyTlsTest.java
@@ -58,7 +58,6 @@ protected void setup() throws Exception {
         proxyConfig.setServicePortTls(PortManager.nextFreePort());
         proxyConfig.setWebServicePort(PortManager.nextFreePort());
         proxyConfig.setWebServicePortTls(PortManager.nextFreePort());
-        proxyConfig.setTlsEnabledInProxy(true);
         proxyConfig.setTlsEnabledWithBroker(false);
         proxyConfig.setTlsCertificateFilePath(TLS_PROXY_CERT_FILE_PATH);
         proxyConfig.setTlsKeyFilePath(TLS_PROXY_KEY_FILE_PATH);
@@ -83,7 +82,7 @@ protected void cleanup() throws Exception {
     @Test
     public void testProducer() throws Exception {
         PulsarClient client = PulsarClient.builder()
-                .serviceUrl("pulsar+ssl://localhost:" + proxyConfig.getServicePortTls()).enableTls(true)
+                .serviceUrl("pulsar+ssl://localhost:" + proxyConfig.getServicePortTls().get()).enableTls(true)
                 .allowTlsInsecureConnection(false).tlsTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH).build();
         Producer<byte[]> producer = client.newProducer(Schema.BYTES).topic("persistent://sample/test/local/topic").create();
 
@@ -97,7 +96,7 @@ public void testProducer() throws Exception {
     @Test
     public void testPartitions() throws Exception {
         PulsarClient client = PulsarClient.builder()
-                .serviceUrl("pulsar+ssl://localhost:" + proxyConfig.getServicePortTls()).enableTls(true)
+                .serviceUrl("pulsar+ssl://localhost:" + proxyConfig.getServicePortTls().get()).enableTls(true)
                 .allowTlsInsecureConnection(false).tlsTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH).build();
         admin.tenants().createTenant("sample", new TenantInfo());
         admin.topics().createPartitionedTopic("persistent://sample/test/local/partitioned-topic", 2);
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationNegTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationNegTest.java
index 5ee0dac24e..1fb9cb4a7a 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationNegTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationNegTest.java
@@ -78,7 +78,6 @@ protected void setup() throws Exception {
         conf.setAuthenticationEnabled(true);
         conf.setAuthorizationEnabled(true);
 
-        conf.setTlsEnabled(true);
         conf.setTlsTrustCertsFilePath(TLS_PROXY_TRUST_CERT_FILE_PATH);
         conf.setTlsCertificateFilePath(TLS_BROKER_CERT_FILE_PATH);
         conf.setTlsKeyFilePath(TLS_BROKER_KEY_FILE_PATH);
@@ -110,7 +109,6 @@ protected void setup() throws Exception {
         proxyConfig.setServicePortTls(PortManager.nextFreePort());
         proxyConfig.setWebServicePort(PortManager.nextFreePort());
         proxyConfig.setWebServicePortTls(PortManager.nextFreePort());
-        proxyConfig.setTlsEnabledInProxy(true);
         proxyConfig.setTlsEnabledWithBroker(true);
 
         // enable tls and auth&auth at proxy
@@ -158,7 +156,7 @@ public void testProxyAuthorization() throws Exception {
         log.info("-- Starting {} test --", methodName);
 
         createAdminClient();
-        final String proxyServiceUrl = "pulsar://localhost:" + proxyConfig.getServicePortTls();
+        final String proxyServiceUrl = "pulsar://localhost:" + proxyConfig.getServicePortTls().get();
         // create a client which connects to proxy over tls and pass authData
         PulsarClient proxyClient = createPulsarClient(proxyServiceUrl);
 
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationTest.java
index d894b0833f..1f743e29c2 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationTest.java
@@ -142,7 +142,6 @@ protected void setup() throws Exception {
         conf.setAuthenticationEnabled(true);
         conf.setAuthorizationEnabled(true);
 
-        conf.setTlsEnabled(true);
         conf.setTlsTrustCertsFilePath(TLS_PROXY_TRUST_CERT_FILE_PATH);
         conf.setTlsCertificateFilePath(TLS_BROKER_CERT_FILE_PATH);
         conf.setTlsKeyFilePath(TLS_BROKER_KEY_FILE_PATH);
@@ -174,7 +173,6 @@ protected void setup() throws Exception {
         proxyConfig.setServicePortTls(PortManager.nextFreePort());
         proxyConfig.setWebServicePort(PortManager.nextFreePort());
         proxyConfig.setWebServicePortTls(PortManager.nextFreePort());
-        proxyConfig.setTlsEnabledInProxy(true);
         proxyConfig.setTlsEnabledWithBroker(true);
 
         // enable tls and auth&auth at proxy
@@ -223,7 +221,7 @@ public void testProxyAuthorization() throws Exception {
 
         startProxy();
         createAdminClient();
-        final String proxyServiceUrl = "pulsar://localhost:" + proxyConfig.getServicePortTls();
+        final String proxyServiceUrl = "pulsar://localhost:" + proxyConfig.getServicePortTls().get();
         // create a client which connects to proxy over tls and pass authData
         PulsarClient proxyClient = createPulsarClient(proxyServiceUrl, PulsarClient.builder());
 
@@ -274,7 +272,7 @@ public void testTlsHostVerificationProxyToClient(boolean hostnameVerificationEna
 
         startProxy();
         createAdminClient();
-        final String proxyServiceUrl = "pulsar://localhost:" + proxyConfig.getServicePortTls();
+        final String proxyServiceUrl = "pulsar://localhost:" + proxyConfig.getServicePortTls().get();
         // create a client which connects to proxy over tls and pass authData
         PulsarClient proxyClient = createPulsarClient(proxyServiceUrl,
                 PulsarClient.builder().enableTlsHostnameVerification(hostnameVerificationEnabled));
@@ -324,7 +322,7 @@ public void testTlsHostVerificationProxyToBroker(boolean hostnameVerificationEna
         proxyConfig.setTlsHostnameVerificationEnabled(hostnameVerificationEnabled);
         startProxy();
         createAdminClient();
-        final String proxyServiceUrl = "pulsar://localhost:" + proxyConfig.getServicePortTls();
+        final String proxyServiceUrl = "pulsar://localhost:" + proxyConfig.getServicePortTls().get();
         // create a client which connects to proxy over tls and pass authData
         PulsarClient proxyClient = createPulsarClient(proxyServiceUrl,
                 PulsarClient.builder().operationTimeout(1, TimeUnit.SECONDS));
@@ -385,7 +383,6 @@ public void tlsCiphersAndProtocols(Set<String> tlsCiphers, Set<String> tlsProtoc
         proxyConfig.setServicePortTls(PortManager.nextFreePort());
         proxyConfig.setWebServicePort(PortManager.nextFreePort());
         proxyConfig.setWebServicePortTls(PortManager.nextFreePort());
-        proxyConfig.setTlsEnabledInProxy(true);
         proxyConfig.setTlsEnabledWithBroker(true);
 
         // enable tls and auth&auth at proxy
@@ -418,7 +415,7 @@ public void tlsCiphersAndProtocols(Set<String> tlsCiphers, Set<String> tlsProtoc
         }, 3, 1000);
         try {
 
-            final String proxyServiceUrl = "pulsar://localhost:" + proxyConfig.getServicePortTls();
+            final String proxyServiceUrl = "pulsar://localhost:" + proxyConfig.getServicePortTls().get();
             PulsarClient proxyClient = createPulsarClient(proxyServiceUrl, PulsarClient.builder());
             Consumer<byte[]> consumer = proxyClient.newConsumer()
                     .topic("persistent://my-property/proxy-authorization/my-ns/my-topic1")
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithoutServiceDiscoveryTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithoutServiceDiscoveryTest.java
index 4909a89d65..825982c3df 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithoutServiceDiscoveryTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithoutServiceDiscoveryTest.java
@@ -70,7 +70,6 @@ protected void setup() throws Exception {
         conf.setAuthenticationEnabled(true);
         conf.setAuthorizationEnabled(false);
 
-        conf.setTlsEnabled(true);
         conf.setTlsTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH);
         conf.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
         conf.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
@@ -102,7 +101,6 @@ protected void setup() throws Exception {
         proxyConfig.setServicePortTls(PortManager.nextFreePort());
         proxyConfig.setWebServicePort(PortManager.nextFreePort());
         proxyConfig.setWebServicePortTls(PortManager.nextFreePort());
-        proxyConfig.setTlsEnabledInProxy(true);
         proxyConfig.setTlsEnabledWithBroker(true);
 
         // enable tls and auth&auth at proxy
@@ -149,7 +147,7 @@ protected void cleanup() throws Exception {
     public void testDiscoveryService() throws Exception {
         log.info("-- Starting {} test --", methodName);
 
-        final String proxyServiceUrl = "pulsar://localhost:" + proxyConfig.getServicePortTls();
+        final String proxyServiceUrl = "pulsar://localhost:" + proxyConfig.getServicePortTls().get();
         Map<String, String> authParams = Maps.newHashMap();
         authParams.put("tlsCertFile", TLS_CLIENT_CERT_FILE_PATH);
         authParams.put("tlsKeyFile", TLS_CLIENT_KEY_FILE_PATH);
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/SuperUserAuthedAdminProxyHandlerTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/SuperUserAuthedAdminProxyHandlerTest.java
index 9d5317c3d6..4bca6d4ca1 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/SuperUserAuthedAdminProxyHandlerTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/SuperUserAuthedAdminProxyHandlerTest.java
@@ -64,7 +64,6 @@ protected void setup() throws Exception {
         conf.setAuthenticationEnabled(true);
         conf.setAuthorizationEnabled(true);
 
-        conf.setTlsEnabled(true);
         conf.setTlsTrustCertsFilePath(getTlsFile("ca.cert"));
         conf.setTlsCertificateFilePath(getTlsFile("broker.cert"));
         conf.setTlsKeyFilePath(getTlsFile("broker.key-pk8"));
@@ -82,7 +81,6 @@ protected void setup() throws Exception {
         proxyConfig.setServicePortTls(PortManager.nextFreePort());
         proxyConfig.setWebServicePort(PortManager.nextFreePort());
         proxyConfig.setWebServicePortTls(PortManager.nextFreePort());
-        proxyConfig.setTlsEnabledInProxy(true);
         proxyConfig.setTlsEnabledWithBroker(true);
 
         // enable tls and auth&auth at proxy
@@ -121,7 +119,7 @@ protected void cleanup() throws Exception {
 
     PulsarAdmin getAdminClient(String user) throws Exception {
         return PulsarAdmin.builder()
-            .serviceHttpUrl("https://localhost:" + proxyConfig.getWebServicePortTls())
+            .serviceHttpUrl("https://localhost:" + proxyConfig.getWebServicePortTls().get())
             .tlsTrustCertsFilePath(getTlsFile("ca.cert"))
             .allowTlsInsecureConnection(false)
             .authentication(AuthenticationTls.class.getName(),
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/UnauthedAdminProxyHandlerTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/UnauthedAdminProxyHandlerTest.java
index 5085d14552..492300538a 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/UnauthedAdminProxyHandlerTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/UnauthedAdminProxyHandlerTest.java
@@ -18,7 +18,6 @@
  */
 package org.apache.pulsar.proxy.server;
 
-import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.spy;
 
 import java.util.HashSet;
@@ -26,7 +25,6 @@
 import java.util.Set;
 
 import javax.servlet.http.HttpServletRequest;
-import javax.validation.constraints.AssertTrue;
 import javax.ws.rs.client.Client;
 import javax.ws.rs.client.ClientBuilder;
 import javax.ws.rs.client.WebTarget;
@@ -102,7 +100,7 @@ protected void cleanup() throws Exception {
     @Test
     public void testUnauthenticatedProxy() throws Exception {
         PulsarAdmin admin = PulsarAdmin.builder()
-            .serviceHttpUrl("http://127.0.0.1:" + proxyConfig.getWebServicePort())
+            .serviceHttpUrl("http://127.0.0.1:" + proxyConfig.getWebServicePort().get())
             .build();
         List<String> activeBrokers = admin.brokers().getActiveBrokers(configClusterName);
         Assert.assertEquals(activeBrokers.size(), 1);
@@ -113,7 +111,7 @@ public void testUnauthenticatedProxy() throws Exception {
     @Test
     public void testVipStatus() throws Exception {
         Client client = ClientBuilder.newClient(new ClientConfig().register(LoggingFeature.class));
-        WebTarget webTarget = client.target("http://127.0.0.1:" + proxyConfig.getWebServicePort())
+        WebTarget webTarget = client.target("http://127.0.0.1:" + proxyConfig.getWebServicePort().get())
                 .path("/status.html");
         String response = webTarget.request().get(String.class);
         Assert.assertEquals(response, "OK");
diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketService.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketService.java
index 6660cd8303..b984f81687 100644
--- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketService.java
+++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketService.java
@@ -190,12 +190,12 @@ private PulsarClient createClientInstance(ClusterData clusterData) throws IOExce
                     config.getBrokerClientAuthenticationParameters());
         }
 
-        if (config.isTlsEnabled()) {
-            if (isNotBlank(clusterData.getBrokerServiceUrlTls())) {
-                clientBuilder.serviceUrl(clusterData.getBrokerServiceUrlTls());
-            } else if (isNotBlank(clusterData.getServiceUrlTls())) {
-                clientBuilder.serviceUrl(clusterData.getServiceUrlTls());
-            }
+        if (config.isBrokerClientTlsEnabled()) {
+			if (isNotBlank(clusterData.getBrokerServiceUrlTls())) {
+					clientBuilder.serviceUrl(clusterData.getBrokerServiceUrlTls());
+			} else if (isNotBlank(clusterData.getServiceUrlTls())) {
+					clientBuilder.serviceUrl(clusterData.getServiceUrlTls());
+			}
         } else if (isNotBlank(clusterData.getBrokerServiceUrl())) {
             clientBuilder.serviceUrl(clusterData.getBrokerServiceUrl());
         } else {
diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/service/ProxyServer.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/service/ProxyServer.java
index c7fd3b1740..147058170c 100644
--- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/service/ProxyServer.java
+++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/service/ProxyServer.java
@@ -65,13 +65,13 @@ public ProxyServer(WebSocketProxyConfiguration config)
         this.server = new Server(executorService);
         List<ServerConnector> connectors = new ArrayList<>();
 
-        ServerConnector connector = new ServerConnector(server);
-
-        connector.setPort(config.getWebServicePort());
-        connectors.add(connector);
-
+        if (config.getWebServicePort().isPresent()) {
+			ServerConnector connector = new ServerConnector(server);
+			connector.setPort(config.getWebServicePort().get());
+			connectors.add(connector);
+        }
         // TLS enabled connector
-        if (config.isTlsEnabled()) {
+        if (config.getWebServicePortTls().isPresent()) {
             try {
                 SslContextFactory sslCtxFactory = SecurityUtility.createSslContextFactory(
                         config.isTlsAllowInsecureConnection(),
@@ -80,12 +80,11 @@ public ProxyServer(WebSocketProxyConfiguration config)
                         config.getTlsKeyFilePath(),
                         config.getTlsRequireTrustedClientCertOnConnect());
                 ServerConnector tlsConnector = new ServerConnector(server, -1, -1, sslCtxFactory);
-                tlsConnector.setPort(config.getWebServicePortTls());
+                tlsConnector.setPort(config.getWebServicePortTls().get());
                 connectors.add(tlsConnector);
             } catch (GeneralSecurityException e) {
                 throw new PulsarServerException(e);
             }
-
         }
 
         // Limit number of concurrent HTTP connections to avoid getting out of
@@ -117,7 +116,7 @@ public void addRestResources(String basePath, String javaPackages, String attrib
     }
 
     public void start() throws PulsarServerException {
-        log.info("Starting web socket proxy at port {}", conf.getWebServicePort());
+        log.info("Starting web socket proxy at port {}", conf.getWebServicePort().get());
         try {
             RequestLogHandler requestLogHandler = new RequestLogHandler();
             Slf4jRequestLog requestLog = new Slf4jRequestLog();
diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/service/WebSocketProxyConfiguration.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/service/WebSocketProxyConfiguration.java
index d89a9a0f07..7f714ae7ac 100644
--- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/service/WebSocketProxyConfiguration.java
+++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/service/WebSocketProxyConfiguration.java
@@ -18,6 +18,7 @@
  */
 package org.apache.pulsar.websocket.service;
 
+import java.util.Optional;
 import java.util.Properties;
 import java.util.Set;
 
@@ -58,9 +59,9 @@
     private long zooKeeperSessionTimeoutMillis = 30000;
 
     // Port to use to server HTTP request
-    private int webServicePort = 8080;
+    private Integer webServicePort = 8080;
     // Port to use to server HTTPS request
-    private int webServicePortTls = 8443;
+    private Integer webServicePortTls;
     // Hostname or IP address the service binds on, default is 0.0.0.0.
     private String bindAddress;
     // --- Authentication ---
@@ -99,8 +100,10 @@
     private String anonymousUserRole = null;
 
     /***** --- TLS --- ****/
-    // Enable TLS
+    @Deprecated 
     private boolean tlsEnabled = false;
+    
+    private boolean brokerClientTlsEnabled = false;
     // Path for the TLS certificate file
     private String tlsCertificateFilePath;
     // Path for the TLS private key file
@@ -189,16 +192,16 @@ public void setZooKeeperSessionTimeoutMillis(long zooKeeperSessionTimeoutMillis)
         this.zooKeeperSessionTimeoutMillis = zooKeeperSessionTimeoutMillis;
     }
 
-    public int getWebServicePort() {
-        return webServicePort;
+    public Optional<Integer> getWebServicePort() {
+        return Optional.ofNullable(webServicePort);
     }
 
     public void setWebServicePort(int webServicePort) {
         this.webServicePort = webServicePort;
     }
 
-    public int getWebServicePortTls() {
-        return webServicePortTls;
+    public Optional<Integer> getWebServicePortTls() {
+        return Optional.ofNullable(webServicePortTls);
     }
 
     public void setWebServicePortTls(int webServicePortTls) {
@@ -317,12 +320,12 @@ public void setAnonymousUserRole(String anonymousUserRole) {
         this.anonymousUserRole = anonymousUserRole;
     }
 
-    public boolean isTlsEnabled() {
-        return tlsEnabled;
+    public boolean isBrokerClientTlsEnabled() {
+        return brokerClientTlsEnabled || tlsEnabled;
     }
 
-    public void setTlsEnabled(boolean tlsEnabled) {
-        this.tlsEnabled = tlsEnabled;
+    public void setBrokerClientTlsEnabled(boolean brokerClientTlsEnabled) {
+        this.brokerClientTlsEnabled = brokerClientTlsEnabled;
     }
 
     public String getTlsCertificateFilePath() {
diff --git a/pulsar-websocket/src/test/java/org/apache/pulsar/websocket/LookupProtocolTest.java b/pulsar-websocket/src/test/java/org/apache/pulsar/websocket/LookupProtocolTest.java
index a6f3845c9b..888ad964d0 100644
--- a/pulsar-websocket/src/test/java/org/apache/pulsar/websocket/LookupProtocolTest.java
+++ b/pulsar-websocket/src/test/java/org/apache/pulsar/websocket/LookupProtocolTest.java
@@ -45,7 +45,6 @@ public void httpLookupTest() throws Exception{
         Field lookupField = PulsarClientImpl.class.getDeclaredField("lookup");
         lookupField.setAccessible(true);
         Assert.assertEquals(lookupField.get(testClient).getClass().getName(), "org.apache.pulsar.client.impl.HttpLookupService");
-        Assert.assertFalse(testClient.getConfiguration().isUseTls());
         service.close();
     }
 
@@ -55,7 +54,7 @@ public void httpsLookupTest() throws Exception{
         conf.setServiceUrl("http://localhost:8080");
         conf.setServiceUrlTls("https://localhost:8443");
         conf.setBrokerServiceUrl("pulsar://localhost:6650");
-        conf.setTlsEnabled(true);
+        conf.setBrokerClientTlsEnabled(true);
         WebSocketService service  = new WebSocketService(conf);
         PulsarClientImpl testClient = (PulsarClientImpl) service.getPulsarClient();
         Field lookupField = PulsarClientImpl.class.getDeclaredField("lookup");
@@ -77,7 +76,6 @@ public void binaryLookupTest() throws Exception{
         Field lookupField = PulsarClientImpl.class.getDeclaredField("lookup");
         lookupField.setAccessible(true);
         Assert.assertEquals(lookupField.get(testClient).getClass().getName(), "org.apache.pulsar.client.impl.BinaryProtoLookupService");
-        Assert.assertFalse(testClient.getConfiguration().isUseTls());
         service.close();
     }
 
@@ -88,7 +86,7 @@ public void binaryTlsLookupTest() throws Exception{
         conf.setServiceUrlTls("https://localhost:8443");
         conf.setBrokerServiceUrl("pulsar://localhost:6650");
         conf.setBrokerServiceUrlTls("pulsar+ssl://localhost:6651");
-        conf.setTlsEnabled(true);
+        conf.setBrokerClientTlsEnabled(true);
         WebSocketService service  = new WebSocketService(conf);
         PulsarClientImpl testClient = (PulsarClientImpl) service.getPulsarClient();
         Field lookupField = PulsarClientImpl.class.getDeclaredField("lookup");


 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services

Mime
View raw message