jai1 closed pull request #1248: Start Proxy in TLS only mode.
URL: https://github.com/apache/incubator-pulsar/pull/1248
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/conf/broker.conf b/conf/broker.conf
index 59a19d7b8b..4eaddefb3f 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -29,13 +29,13 @@ globalZookeeperServers=
brokerServicePort=6650
# Broker data port for TLS
-brokerServicePortTls=6651
+brokerServicePortTls=
# Port to use to server HTTP request
webServicePort=8080
# Port to use to server HTTPS request
-webServicePortTls=8443
+webServicePortTls=
# Hostname or IP address the service binds on, default is 0.0.0.0.
bindAddress=0.0.0.0
@@ -192,9 +192,6 @@ proxyRoles=
# else it just accepts the originalPrincipal and authorizes it (if required).
authenticateOriginalAuthData=false
-# Enable TLS
-tlsEnabled=false
-
# Path for the TLS certificate file
tlsCertificateFilePath=
diff --git a/conf/discovery.conf b/conf/discovery.conf
index 49f499a080..4957d9429d 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,8 +65,6 @@ superUserRoles=
authorizationAllowWildcardsMatching=false
##### --- TLS --- #####
-# Enable TLS
-tlsEnabled=false
# Path for the TLS certificate file
tlsCertificateFilePath=
diff --git a/conf/proxy.conf b/conf/proxy.conf
index 384cca06b6..aa8626be8d 100644
--- a/conf/proxy.conf
+++ b/conf/proxy.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=
# Path for the file used to determine the rotation status for the
proxy-instance when responding
# to service discovery health checks
@@ -71,9 +71,6 @@ forwardAuthorizationCredentials=false
##### --- TLS --- #####
-# Enable TLS in the proxy
-tlsEnabledInProxy=false
-
# Enable TLS when talking with the brokers
tlsEnabledWithBroker=false
diff --git a/conf/websocket.conf b/conf/websocket.conf
index 0ceda6273d..e720a5678a 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
@@ -85,9 +85,6 @@ anonymousUserRole=
### --- TLS --- ###
-# Enable TLS
-tlsEnabled=false
-
# Accept untrusted TLS certificate from client
tlsAllowInsecureConnection=false
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 f851f7dd93..aa0972383a 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
@@ -43,12 +43,12 @@
// Global Zookeeper quorum connection string
@FieldContext(required = false)
private String globalZookeeperServers;
- private int brokerServicePort = 6650;
- private int brokerServicePortTls = 6651;
+ private Integer brokerServicePort = 6650;
+ private Integer brokerServicePortTls;
// 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.
private String bindAddress = "0.0.0.0";
@@ -186,8 +186,6 @@
private int maxConsumersPerSubscription = 0;
/***** --- TLS --- ****/
- // Enable TLS
- private boolean tlsEnabled = false;
// Path for the TLS certificate file
private String tlsCertificateFilePath;
// Path for the TLS private key file
@@ -457,32 +455,32 @@ public void setGlobalZookeeperServers(String
globalZookeeperServers) {
this.globalZookeeperServers = globalZookeeperServers;
}
- public int getBrokerServicePort() {
- return brokerServicePort;
+ public Optional<Integer> getBrokerServicePort() {
+ return Optional.ofNullable(brokerServicePort);
}
public void setBrokerServicePort(int brokerServicePort) {
this.brokerServicePort = brokerServicePort;
}
- public int getBrokerServicePortTls() {
- return brokerServicePortTls;
+ public Optional<Integer> getBrokerServicePortTls() {
+ return Optional.ofNullable(brokerServicePortTls);
}
public void setBrokerServicePortTls(int brokerServicePortTls) {
this.brokerServicePortTls = brokerServicePortTls;
}
- 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) {
@@ -795,14 +793,6 @@ public void setMaxConsumersPerSubscription(int
maxConsumersPerSubscription) {
this.maxConsumersPerSubscription = maxConsumersPerSubscription;
}
- public boolean isTlsEnabled() {
- return tlsEnabled;
- }
-
- public void setTlsEnabled(boolean tlsEnabled) {
- this.tlsEnabled = tlsEnabled;
- }
-
public String getTlsCertificateFilePath() {
return tlsCertificateFilePath;
}
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 038fb4dec3..9a23fc4bae 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.getGlobalZookeeperServers(),
"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/broker/PulsarService.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
index 98a4e08e57..e490ded2e5 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
@@ -665,24 +665,32 @@ public static String
advertisedAddress(ServiceConfiguration config) {
}
public static String brokerUrl(ServiceConfiguration config) {
- return "pulsar://" + advertisedAddress(config) + ":" +
config.getBrokerServicePort();
+ if (config.getBrokerServicePort().isPresent()) {
+ return "pulsar://" + advertisedAddress(config) + ":" +
config.getBrokerServicePort().get();
+ } else {
+ return "";
+ }
}
public static String brokerUrlTls(ServiceConfiguration config) {
- if (config.isTlsEnabled()) {
- return "pulsar://" + advertisedAddress(config) + ":" +
config.getBrokerServicePortTls();
+ if (config.getBrokerServicePortTls().isPresent()) {
+ return "pulsar://" + advertisedAddress(config) + ":" +
config.getBrokerServicePortTls().get();
} else {
return "";
}
}
public static String webAddress(ServiceConfiguration config) {
- return String.format("http://%s:%d";, advertisedAddress(config),
config.getWebServicePort());
+ if (config.getWebServicePort().isPresent()) {
+ return String.format("http://%s:%d";, advertisedAddress(config),
config.getWebServicePort().get());
+ } else {
+ return "";
+ }
}
public static String webAddressTls(ServiceConfiguration config) {
- if (config.isTlsEnabled()) {
- return String.format("https://%s:%d";, advertisedAddress(config),
config.getWebServicePortTls());
+ if (config.getWebServicePortTls().isPresent()) {
+ return String.format("https://%s:%d";, advertisedAddress(config),
config.getWebServicePortTls().get());
} else {
return "";
}
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java
index 60329d7670..ee75d10b89 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java
@@ -68,7 +68,8 @@
try {
// Add Native brokers
- return
pulsar().getLocalZkCache().getChildren(LoadManager.LOADBALANCE_BROKERS_ROOT);
+ Set<String> te =
pulsar().getLocalZkCache().getChildren(LoadManager.LOADBALANCE_BROKERS_ROOT);
+ return te;
} catch (Exception e) {
LOG.error(String.format("[%s] Failed to get active broker list:
cluster=%s", clientAppId(), cluster), e);
throw new RestException(e);
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 594b717625..1168ed6343 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
@@ -776,11 +776,31 @@ 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;
+ if (config.getWebServicePort().isPresent()) {
+ port = config.getWebServicePort().get();
+ } else if (config.getWebServicePortTls().isPresent()) {
+ port = config.getWebServicePortTls().get();
+ } else if (config.getBrokerServicePort().isPresent()) {
+ port = config.getBrokerServicePort().get();
+ } else {
+ port = config.getBrokerServicePortTls().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());
+ Integer port;
+ if (config.getWebServicePort().isPresent()) {
+ port = config.getWebServicePort().get();
+ } else if (config.getWebServicePortTls().isPresent()) {
+ port = config.getWebServicePortTls().get();
+ } else if (config.getBrokerServicePort().isPresent()) {
+ port = config.getBrokerServicePort().get();
+ } else {
+ port = config.getBrokerServicePortTls().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 577fab32e7..d8d17f8839 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
@@ -49,7 +49,6 @@
import java.util.function.Consumer;
import java.util.function.Predicate;
-import org.apache.bookkeeper.client.BookKeeper.DigestType;
import org.apache.bookkeeper.common.util.OrderedScheduler;
import org.apache.bookkeeper.mledger.AsyncCallbacks.OpenLedgerCallback;
import org.apache.bookkeeper.mledger.ManagedLedger;
@@ -57,6 +56,7 @@
import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.bookkeeper.mledger.ManagedLedgerFactory;
import org.apache.bookkeeper.util.ZkUtils;
+import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.pulsar.broker.PulsarService;
@@ -131,8 +131,6 @@
private final PulsarService pulsar;
private final ManagedLedgerFactory managedLedgerFactory;
- private final int port;
- private final int tlsPort;
private final ConcurrentOpenHashMap<String, CompletableFuture<Topic>>
topics;
@@ -187,8 +185,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();
@@ -284,11 +280,15 @@ public void start() throws Exception {
ServiceConfiguration serviceConfig = pulsar.getConfiguration();
bootstrap.childHandler(new PulsarChannelInitializer(this,
serviceConfig, false));
- // Bind and start to accept incoming connections.
- bootstrap.bind(new InetSocketAddress(pulsar.getBindAddress(),
port)).sync();
- log.info("Started Pulsar Broker service on port {}", port);
-
- if (serviceConfig.isTlsEnabled()) {
+ if (StringUtils.isNotBlank(pulsar.getBrokerServiceUrl())) {
+ final int port = new URI(pulsar.getBrokerServiceUrl()).getPort();
+ // Bind and start to accept incoming connections.
+ bootstrap.bind(new InetSocketAddress(pulsar.getBindAddress(),
port)).sync();
+ log.info("Started Pulsar Broker service on port {}", port);
+ }
+
+ if (StringUtils.isNotBlank(pulsar.getBrokerServiceUrlTls())) {
+ final int tlsPort = new
URI(pulsar.getBrokerServiceUrlTls()).getPort();
ServerBootstrap tlsBootstrap = bootstrap.clone();
tlsBootstrap.childHandler(new PulsarChannelInitializer(this,
serviceConfig, true));
tlsBootstrap.bind(new InetSocketAddress(pulsar.getBindAddress(),
tlsPort)).sync();
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 bae1019403..7bf4676c78 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
@@ -264,7 +264,8 @@ protected void validateClusterOwnership(String cluster)
throws WebApplicationExc
private URI getRedirectionUrl(ClusterData differentClusterData) throws
MalformedURLException {
URL webUrl = null;
- if (pulsar.getConfiguration().isTlsEnabled() &&
!differentClusterData.getServiceUrlTls().isEmpty()) {
+ if (pulsar.getConfiguration().getBrokerServicePortTls().isPresent()
+ && !differentClusterData.getServiceUrlTls().isEmpty()) {
webUrl = new URL(differentClusterData.getServiceUrlTls());
} else {
webUrl = new URL(differentClusterData.getServiceUrl());
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 02c8b1a878..6d500b81db 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
@@ -84,28 +84,29 @@ public WebService(PulsarService pulsar) throws
PulsarServerException {
this.server = new Server(new ExecutorThreadPool(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);
-
- if (pulsar.getConfiguration().isTlsEnabled()) {
+ if (pulsar.getConfiguration().getWebServicePort().isPresent()) {
+ ServerConnector connector = new PulsarServerConnector(server, 1,
1);
+
connector.setPort(pulsar.getConfiguration().getWebServicePort().get());
+ connector.setHost(pulsar.getBindAddress());
+ connectors.add(connector);
+ }
+
+ if (pulsar.getConfiguration().getWebServicePortTls().isPresent()) {
SslContextFactory sslCtxFactory = new SslContextFactory();
try {
sslCtxFactory.setSslContext(
- SecurityUtility.createSslContext(
-
pulsar.getConfiguration().isTlsAllowInsecureConnection(),
-
pulsar.getConfiguration().getTlsTrustCertsFilePath(),
-
pulsar.getConfiguration().getTlsCertificateFilePath(),
- pulsar.getConfiguration().getTlsKeyFilePath()));
+
SecurityUtility.createSslContext(pulsar.getConfiguration().isTlsAllowInsecureConnection(),
+
pulsar.getConfiguration().getTlsTrustCertsFilePath(),
+
pulsar.getConfiguration().getTlsCertificateFilePath(),
+
pulsar.getConfiguration().getTlsKeyFilePath()));
} catch (GeneralSecurityException e) {
throw new PulsarServerException(e);
}
sslCtxFactory.setWantClientAuth(true);
ServerConnector tlsConnector = new PulsarServerConnector(server,
1, 1, sslCtxFactory);
-
tlsConnector.setPort(pulsar.getConfiguration().getWebServicePortTls());
+
tlsConnector.setPort(pulsar.getConfiguration().getWebServicePortTls().get());
tlsConnector.setHost(pulsar.getBindAddress());
connectors.add(tlsConnector);
}
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 073614623b..f3c2aa70b0 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
@@ -81,17 +81,21 @@ public static void main(String[] args) throws Exception {
arguments.brokerConfigFile, ServiceConfiguration.class);
}
- String pulsarServiceUrl = PulsarService.brokerUrl(brokerConfig);
+ String pulsarServiceUrl;
ClientConfiguration clientConfig = new ClientConfiguration();
if (isNotBlank(brokerConfig.getBrokerClientAuthenticationPlugin())) {
clientConfig.setAuthentication(brokerConfig.getBrokerClientAuthenticationPlugin(),
brokerConfig.getBrokerClientAuthenticationParameters());
}
- clientConfig.setUseTls(brokerConfig.isTlsEnabled());
-
clientConfig.setTlsAllowInsecureConnection(brokerConfig.isTlsAllowInsecureConnection());
-
clientConfig.setTlsTrustCertsFilePath(brokerConfig.getTlsCertificateFilePath());
-
+ if (brokerConfig.getBrokerServicePortTls().isPresent()) {
+ pulsarServiceUrl = PulsarService.brokerUrlTls(brokerConfig);
+ clientConfig.setUseTls(true);
+
clientConfig.setTlsAllowInsecureConnection(brokerConfig.isTlsAllowInsecureConnection());
+
clientConfig.setTlsTrustCertsFilePath(brokerConfig.getTlsCertificateFilePath());
+ } else {
+ pulsarServiceUrl = 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/admin/AdminApiTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
index fe89a327e1..95b5658cde 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
@@ -126,7 +126,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/service/BrokerServiceTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
index c8dc63a858..9449c06886 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
@@ -406,7 +406,6 @@ public void testTlsDisabled() throws Exception {
PulsarClient pulsarClient = null;
conf.setAuthenticationEnabled(false);
- conf.setTlsEnabled(false);
restartBroker();
// Case 1: Access without TLS
@@ -445,7 +444,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();
@@ -523,7 +521,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);
@@ -582,7 +579,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);
@@ -640,7 +636,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/PeerReplicatorTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PeerReplicatorTest.java
index d23356e2b0..ce3e26d900 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PeerReplicatorTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PeerReplicatorTest.java
@@ -89,10 +89,6 @@ public void testPeerClusterTopicLookup(String protocol)
throws Exception {
admin1.namespaces().setNamespaceReplicationClusters(namespace1,
Lists.newArrayList("r1"));
admin1.namespaces().setNamespaceReplicationClusters(namespace2,
Lists.newArrayList("r2"));
admin1.clusters().updatePeerClusterNames("r3", null);
- // disable tls as redirection url is prepared according tls
configuration
- pulsar1.getConfiguration().setTlsEnabled(false);
- pulsar2.getConfiguration().setTlsEnabled(false);
- pulsar3.getConfiguration().setTlsEnabled(false);
final String topic1 = "persistent://" + namespace1 + "/topic1";
final String topic2 = "persistent://" + namespace2 + "/topic2";
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 abde142559..bcbc0d33c5 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);
@@ -193,7 +191,6 @@ void setup() throws Exception {
inSec(getBrokerServicePurgeInactiveFrequency(),
TimeUnit.SECONDS));
config3.setBrokerServicePort(PortManager.nextFreePort());
config3.setBrokerServicePortTls(PortManager.nextFreePort());
- config3.setTlsEnabled(true);
config3.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
config3.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
config3.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 c7092de710..a369c6a37e 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
@@ -250,7 +250,6 @@ private void setupEnv(boolean enableFilter, String
minApiVersion, boolean allowU
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 eafe9c8fcd..80a49fcf0e 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
@@ -66,7 +66,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 4338ae32da..7f312cab3a 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
@@ -73,7 +73,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 57d8b478b4..87223475e1 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
@@ -392,7 +392,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());
@@ -401,7 +400,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();
@@ -529,7 +527,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();
@@ -539,7 +536,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);
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 66b2265f20..0c98ba5684 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.setClusterName(clusterName);
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/websocket/proxy/ProxyPublishConsumeTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java
index 3c401733ad..c58ae4ccfc 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
@@ -404,7 +404,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/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 ac79c8a000..5cab868eab 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
@@ -64,7 +64,6 @@ public void setup() throws Exception {
WebSocketProxyConfiguration config = new WebSocketProxyConfiguration();
config.setWebServicePort(port);
config.setWebServicePortTls(tlsPort);
- config.setTlsEnabled(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 38ae7eb366..5f0a55fd6d 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
@@ -36,6 +36,8 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.common.base.Preconditions;
+
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.AdaptiveRecvByteBufAllocator;
@@ -102,16 +104,20 @@ public void startServer() throws Exception {
bootstrap.channel(EventLoopUtil.getServerSocketChannelClass(workerGroup));
EventLoopUtil.enableTriggeredMode(bootstrap);
+ Preconditions.checkArgument(config.getServicePort().isPresent() ||
config.getServicePortTls().isPresent(),
+ "Either ServicePort or ServicePortTls should be configured.");
+
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()) {
+ 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());
}
}
@@ -153,12 +159,17 @@ 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 "";
+ }
}
public String serviceUrlTls() {
- if (config.isTlsEnabled()) {
- return new
StringBuilder("pulsar://").append(host()).append(":").append(config.getServicePortTls())
+ if (config.getServicePortTls().isPresent()) {
+ return new
StringBuilder("pulsar://").append(host()).append(":").append(config.getServicePortTls().get())
.toString();
} else {
return "";
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 d8be507490..340272970e 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
@@ -58,20 +58,19 @@
private final Server server;
private final ExecutorService webServiceExecutor;
private final List<Handler> handlers = Lists.newArrayList();
- protected final int externalServicePort;
public ServerManager(ServiceConfig config) {
this.webServiceExecutor = Executors.newFixedThreadPool(32, new
DefaultThreadFactory("pulsar-external-web"));
this.server = new Server(new ExecutorThreadPool(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.isTlsEnabled()) {
+ if (config.getWebServicePort().isPresent()) {
+ ServerConnector connector = new ServerConnector(server, 1, 1);
+ connector.setPort(config.getWebServicePort().get());
+ connectors.add(connector);
+ }
+ if (config.getWebServicePortTls().isPresent()) {
SslContextFactory sslCtxFactory = new SslContextFactory();
try {
SSLContext sslCtx =
SecurityUtility.createSslContext(config.isTlsAllowInsecureConnection(),
config.getTlsTrustCertsFilePath(), config.getTlsCertificateFilePath(),
@@ -83,7 +82,7 @@ public ServerManager(ServiceConfig config) {
sslCtxFactory.setWantClientAuth(true);
ServerConnector tlsConnector = new ServerConnector(server, 1, 1,
sslCtxFactory);
- tlsConnector.setPort(config.getWebServicePortTls());
+ tlsConnector.setPort(config.getWebServicePortTls().get());
connectors.add(tlsConnector);
}
@@ -106,10 +105,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 8cf56d11ac..c25552b345 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;
@@ -42,13 +43,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;
@@ -72,8 +73,6 @@
private String authorizationProvider =
PulsarAuthorizationProvider.class.getName();
/***** --- TLS --- ****/
- // Enable TLS
- private boolean tlsEnabled = false;
// Path for the TLS certificate file
private String tlsCertificateFilePath;
// Path for the TLS private key file
@@ -115,46 +114,38 @@ 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;
}
- public boolean isTlsEnabled() {
- return tlsEnabled;
- }
-
- public void setTlsEnabled(boolean tlsEnabled) {
- this.tlsEnabled = tlsEnabled;
- }
-
public String getTlsCertificateFilePath() {
return tlsCertificateFilePath;
}
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-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 69329efc06..08910bb2e8 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
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.proxy.server;
+import java.util.Optional;
import java.util.Properties;
import java.util.Set;
@@ -43,12 +44,12 @@
// Port to use to server binary-proto request
private int servicePort = 6650;
// Port to use to server binary-proto-tls request
- private int servicePortTls = 6651;
+ private int servicePortTls;
// Port to use to server HTTP request
private int webServicePort = 8080;
// Port to use to server HTTPS request
- private int webServicePortTls = 8443;
+ private int webServicePortTls;
// Path for the file used to determine the rotation status for the broker
// when responding to service discovery health checks
@@ -76,9 +77,6 @@
private String brokerClientTrustCertsFilePath;
/***** --- TLS --- ****/
- // Enable TLS for the proxy handler
- private boolean tlsEnabledInProxy = false;
-
// Enable TLS when talking with the brokers
private boolean tlsEnabledWithBroker = false;
@@ -149,32 +147,32 @@ 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) {
@@ -189,14 +187,6 @@ public void setStatusFilePath(String statusFilePath) {
this.statusFilePath = statusFilePath;
}
- public boolean isTlsEnabledInProxy() {
- return tlsEnabledInProxy;
- }
-
- public void setTlsEnabledInProxy(boolean tlsEnabledInProxy) {
- this.tlsEnabledInProxy = tlsEnabledInProxy;
- }
-
public boolean isTlsEnabledWithBroker() {
return tlsEnabledWithBroker;
}
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 433744273e..3e467b7832 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
@@ -26,6 +26,7 @@
import java.net.InetAddress;
import java.net.UnknownHostException;
+import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.authentication.AuthenticationService;
import org.apache.pulsar.broker.authorization.AuthorizationService;
@@ -44,6 +45,8 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.common.base.Preconditions;
+
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.AdaptiveRecvByteBufAllocator;
@@ -90,22 +93,36 @@ public ProxyService(ProxyConfiguration proxyConfig) throws
IOException {
} 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());
+
+ Preconditions.checkArgument(
+ proxyConfig.getServicePort().isPresent() ||
proxyConfig.getServicePortTls().isPresent(),
+ "Need to define the Service ports for proxy.");
+
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(proxyConfig.getBrokerServiceURLTLS())
+ ||
StringUtils.isNotBlank(proxyConfig.getBrokerServiceURL()),
+ "Need to specify brokerServiceURL or brokerServiceURLTLS.");
+
+ this.serviceUrl = proxyConfig.getServicePort().isPresent()
+ ? String.format("pulsar://%s:%d/", hostname,
proxyConfig.getServicePort()) : null;
+ this.serviceUrlTls = proxyConfig.getServicePortTls().isPresent()
+ ? String.format("pulsar://%s:%d/", hostname,
proxyConfig.getServicePortTls()) : null;
this.acceptorGroup = EventLoopUtil.newEventLoopGroup(1,
acceptorThreadFactory);
this.workerGroup = EventLoopUtil.newEventLoopGroup(numThreads,
workersThreadFactory);
ClientConfigurationData clientConf = new ClientConfigurationData();
- clientConf.setServiceUrl(serviceUrl);
- if (proxyConfig.getBrokerClientAuthenticationPlugin() != null) {
+ if
(StringUtils.isNotBlank(proxyConfig.getBrokerClientAuthenticationPlugin())) {
clientConf.setAuthentication(AuthenticationFactory.create(proxyConfig.getBrokerClientAuthenticationPlugin(),
proxyConfig.getBrokerClientAuthenticationParameters()));
}
- if (proxyConfig.isTlsEnabledWithBroker()) {
+ if (StringUtils.isNotBlank(proxyConfig.getBrokerServiceURLTLS())) {
+ clientConf.setServiceUrl(proxyConfig.getBrokerServiceURLTLS());
clientConf.setUseTls(true);
clientConf.setTlsTrustCertsFilePath(proxyConfig.getBrokerClientTrustCertsFilePath());
clientConf.setTlsAllowInsecureConnection(proxyConfig.isTlsAllowInsecureConnection());
+ } else {
+ clientConf.setServiceUrl(proxyConfig.getBrokerServiceURL());
}
this.client = new PulsarClientImpl(clientConf, workerGroup);
@@ -143,14 +160,15 @@ public void shutdown(int exitCode) {
bootstrap.childHandler(new ServiceChannelInitializer(this,
proxyConfig, false));
// Bind and start to accept incoming connections.
- bootstrap.bind(proxyConfig.getServicePort()).sync();
- LOG.info("Started Pulsar Proxy at {}", serviceUrl);
-
- if (proxyConfig.isTlsEnabledInProxy()) {
+ if (proxyConfig.getServicePort().isPresent()) {
+ bootstrap.bind(proxyConfig.getServicePort().get()).sync();
+ LOG.info("Started Pulsar Proxy at {}", serviceUrl);
+ }
+ 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 at {}", serviceUrlTls);
}
}
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 edc7188288..baaabc15a5 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
@@ -59,20 +59,20 @@
private final Server server;
private final ExecutorService webServiceExecutor;
private final List<Handler> handlers = Lists.newArrayList();
- protected final int externalServicePort;
public WebServer(ProxyConfiguration config) {
this.webServiceExecutor = Executors.newFixedThreadPool(32, new
DefaultThreadFactory("pulsar-external-web"));
this.server = new Server(new ExecutorThreadPool(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()) {
+ connector.setPort(config.getWebServicePort().get());
+ connectors.add(connector);
+ }
- if (config.isTlsEnabledInProxy()) {
+ if (config.getWebServicePortTls().isPresent()) {
SslContextFactory sslCtxFactory = new SslContextFactory();
try {
SSLContext sslCtx = SecurityUtility.createSslContext(false,
null, config.getTlsCertificateFilePath(),
@@ -84,7 +84,7 @@ public WebServer(ProxyConfiguration config) {
sslCtxFactory.setWantClientAuth(false);
ServerConnector tlsConnector = new ServerConnector(server, 1, 1,
sslCtxFactory);
- tlsConnector.setPort(config.getWebServicePortTls());
+ tlsConnector.setPort(config.getWebServicePortTls().get());
connectors.add(tlsConnector);
}
@@ -117,10 +117,6 @@ public void addRestResources(String basePath, String
javaPackages, String attrib
context.setAttribute(attribute, attributeValue);
handlers.add(context);
}
-
- public int getExternalServicePort() {
- return externalServicePort;
- }
public void start() throws Exception {
RequestLogHandler requestLogHandler = new RequestLogHandler();
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 4b3a914e1a..16ea8f8a1e 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
@@ -72,7 +72,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);
@@ -103,7 +102,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
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 c06e258752..5db44fd27e 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
@@ -53,7 +53,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/ProxyRolesEnforcementTest.java
b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyRolesEnforcementTest.java
index 3c4361b178..289abbfb31 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");
@@ -253,7 +252,7 @@ private void createAdminClient() throws
PulsarClientException {
}
private PulsarClient createPulsarClient(String proxyServiceUrl, String
authParams) throws PulsarClientException {
- return
PulsarClient.builder().serviceUrl(proxyServiceUrl).authentication(BasicAuthentication.class.getName(),
- authParams).build();
+ return PulsarClient.builder().serviceUrl(proxyServiceUrl)
+ .authentication(BasicAuthentication.class.getName(),
authParams).build();
}
}
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 510c48d7a2..ac0db12b13 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
@@ -55,7 +55,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);
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 08c6bc0c62..af91d68f40 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
@@ -76,7 +76,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);
@@ -108,7 +107,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
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 5ccbe8e02e..86a85ea4d2 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
@@ -140,7 +140,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);
@@ -172,7 +171,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
@@ -382,7 +380,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
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 834410921c..2369775c72 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
@@ -68,7 +68,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);
@@ -100,7 +99,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
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 b917dc2447..3e6a256ae4 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
@@ -31,6 +31,7 @@
import javax.websocket.DeploymentException;
import org.apache.bookkeeper.common.util.OrderedScheduler;
+import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.broker.PulsarServerException;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.authentication.AuthenticationService;
@@ -82,7 +83,7 @@
private final ConcurrentOpenHashMap<String,
ConcurrentOpenHashSet<ConsumerHandler>> topicConsumerMap;
private final ConcurrentOpenHashMap<String,
ConcurrentOpenHashSet<ReaderHandler>> topicReaderMap;
private final ProxyStats proxyStats;
-
+
public WebSocketService(WebSocketProxyConfiguration config) {
this(createClusterData(config),
PulsarConfigurationLoader.convertFrom(config));
}
@@ -178,7 +179,8 @@ public synchronized PulsarClient getPulsarClient() throws
IOException {
private PulsarClient createClientInstance(ClusterData clusterData) throws
IOException {
ClientBuilder clientBuilder = PulsarClient.builder() //
.statsInterval(0, TimeUnit.SECONDS) //
- .enableTls(config.isTlsEnabled()) //
+
.enableTls(StringUtils.isNotBlank(clusterData.getBrokerServiceUrlTls())
+ ||
StringUtils.isNotBlank(clusterData.getServiceUrlTls())) //
.allowTlsInsecureConnection(config.isTlsAllowInsecureConnection()) //
.tlsTrustCertsFilePath(config.getBrokerClientTrustCertsFilePath()) //
.ioThreads(config.getWebSocketNumIoThreads()) //
@@ -190,12 +192,10 @@ 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 (StringUtils.isNotBlank(clusterData.getBrokerServiceUrlTls())) {
+ clientBuilder.serviceUrl(clusterData.getBrokerServiceUrlTls());
+ } else if (StringUtils.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 77b24c143f..d590416d7d 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
@@ -74,11 +74,12 @@ public ProxyServer(WebSocketProxyConfiguration config)
ServerConnector connector = new ServerConnector(server);
- connector.setPort(config.getWebServicePort());
- connectors.add(connector);
-
+ if (config.getWebServicePort().isPresent()) {
+ connector.setPort(config.getWebServicePort().get());
+ connectors.add(connector);
+ }
// TLS enabled connector
- if (config.isTlsEnabled()) {
+ if (config.getWebServicePortTls().isPresent()) {
SslContextFactory sslCtxFactory = new SslContextFactory(true);
try {
SSLContext sslCtx = SecurityUtility.createSslContext(false,
config.getTlsTrustCertsFilePath(), config.getTlsCertificateFilePath(),
@@ -91,7 +92,7 @@ public ProxyServer(WebSocketProxyConfiguration config)
sslCtxFactory.setWantClientAuth(true);
ServerConnector tlsConnector = new ServerConnector(server, -1, -1,
sslCtxFactory);
- tlsConnector.setPort(config.getWebServicePortTls());
+ tlsConnector.setPort(config.getWebServicePortTls().get());
connectors.add(tlsConnector);
}
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 c3040df66c..5fe376c1ab 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;
@@ -56,9 +57,9 @@
private long zooKeeperSessionTimeoutMillis = 30000;
// Port to use to server HTTP request
- private int webServicePort = 8080;
+ private Integer webServicePort;
// 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 ---
@@ -95,8 +96,6 @@
private String anonymousUserRole = null;
/***** --- TLS --- ****/
- // Enable TLS
- private boolean tlsEnabled = false;
// Path for the TLS certificate file
private String tlsCertificateFilePath;
// Path for the TLS private key file
@@ -172,16 +171,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) {
@@ -292,14 +291,6 @@ public void setAnonymousUserRole(String anonymousUserRole)
{
this.anonymousUserRole = anonymousUserRole;
}
- public boolean isTlsEnabled() {
- return tlsEnabled;
- }
-
- public void setTlsEnabled(boolean tlsEnabled) {
- this.tlsEnabled = tlsEnabled;
- }
-
public String getTlsCertificateFilePath() {
return tlsCertificateFilePath;
}
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..15c0df1855 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
@@ -39,7 +39,6 @@
public void httpLookupTest() throws Exception{
WebSocketProxyConfiguration conf = new WebSocketProxyConfiguration();
conf.setServiceUrl("http://localhost:8080";);
- conf.setServiceUrlTls("https://localhost:8443";);
WebSocketService service = new WebSocketService(conf);
PulsarClientImpl testClient = (PulsarClientImpl)
service.getPulsarClient();
Field lookupField = PulsarClientImpl.class.getDeclaredField("lookup");
@@ -55,7 +54,6 @@ public void httpsLookupTest() throws Exception{
conf.setServiceUrl("http://localhost:8080";);
conf.setServiceUrlTls("https://localhost:8443";);
conf.setBrokerServiceUrl("pulsar://localhost:6650");
- conf.setTlsEnabled(true);
WebSocketService service = new WebSocketService(conf);
PulsarClientImpl testClient = (PulsarClientImpl)
service.getPulsarClient();
Field lookupField = PulsarClientImpl.class.getDeclaredField("lookup");
@@ -69,9 +67,7 @@ public void httpsLookupTest() throws Exception{
public void binaryLookupTest() throws Exception{
WebSocketProxyConfiguration conf = new WebSocketProxyConfiguration();
conf.setServiceUrl("http://localhost:8080";);
- conf.setServiceUrlTls("https://localhost:8443";);
conf.setBrokerServiceUrl("pulsar://localhost:6650");
- conf.setBrokerServiceUrlTls("pulsar+ssl://localhost:6651");
WebSocketService service = new WebSocketService(conf);
PulsarClientImpl testClient = (PulsarClientImpl)
service.getPulsarClient();
Field lookupField = PulsarClientImpl.class.getDeclaredField("lookup");
@@ -88,7 +84,6 @@ public void binaryTlsLookupTest() throws Exception{
conf.setServiceUrlTls("https://localhost:8443";);
conf.setBrokerServiceUrl("pulsar://localhost:6650");
conf.setBrokerServiceUrlTls("pulsar+ssl://localhost:6651");
- conf.setTlsEnabled(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:
[email protected]
With regards,
Apache Git Services
