diff --git a/conf/functions_worker.yml b/conf/functions_worker.yml index 4254d3fa8cbf3..2144e9bf664e4 100644 --- a/conf/functions_worker.yml +++ b/conf/functions_worker.yml @@ -23,6 +23,7 @@ workerId: standalone workerHostname: localhost +# Set to null to disable the non-TLS service port. workerPort: 6750 workerPortTls: 6751 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 fc5feedc5e1ef..aab1f6cbe41bb 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 @@ -1641,10 +1641,8 @@ public static WorkerConfig initializeWorkerConfigFromBrokerConfig(ServiceConfigu String workerConfigFile) throws IOException { WorkerConfig workerConfig = WorkerConfig.load(workerConfigFile); - brokerConfig.getWebServicePort() - .map(port -> workerConfig.setWorkerPort(port)); - brokerConfig.getWebServicePortTls() - .map(port -> workerConfig.setWorkerPortTls(port)); + workerConfig.setWorkerPort(brokerConfig.getWebServicePort().orElse(null)); + workerConfig.setWorkerPortTls(brokerConfig.getWebServicePortTls().orElse(null)); // worker talks to local broker String hostname = ServiceConfigurationUtils.getDefaultOrConfiguredAddress( 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 15ee27dc3a56c..c21b7ea6566ec 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 @@ -23,6 +23,7 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.when; +import static org.testng.Assert.assertFalse; import static org.testng.Assert.assertTrue; import static org.testng.Assert.fail; import com.google.common.collect.Sets; @@ -201,7 +202,7 @@ private PulsarWorkerService createPulsarFunctionWorker(ServiceConfiguration conf workerConfig.setFunctionAssignmentTopicName("assignment"); workerConfig.setFunctionMetadataTopicName("metadata"); workerConfig.setInstanceLivenessCheckFreqMs(100); - workerConfig.setWorkerPort(0); + workerConfig.setWorkerPort(null); workerConfig.setPulsarFunctionsCluster(config.getClusterName()); String hostname = ServiceConfigurationUtils.getDefaultOrConfiguredAddress(config.getAdvertisedAddress()); this.workerId = "c-" + config.getClusterName() + "-fw-" + hostname + "-" + workerConfig.getWorkerPort(); @@ -238,6 +239,11 @@ public PulsarClient newPulsarClient(String pulsarServiceUrl, WorkerConfig worker return workerService; } + @Test + public void testWorkerServerNonTlsNotStarted() { + assertFalse(workerServer.getListenPortHTTP().isPresent(), "Non-TLS worker service should not be set."); + } + @Test public void testAuthorization() { diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java index 7fe566dae7cd3..47c3600fa655d 100644 --- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java +++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java @@ -100,7 +100,7 @@ public class WorkerConfig implements Serializable, PulsarConfiguration { private String workerHostname; @FieldContext( category = CATEGORY_WORKER, - doc = "The port for serving worker http requests" + doc = "The port for serving worker http requests. Set to null to disable serving on the http port." ) private Integer workerPort; @FieldContext( @@ -724,7 +724,8 @@ public static WorkerConfig load(String yamlFile) throws IOException { public String getWorkerId() { if (isBlank(this.workerId)) { - this.workerId = String.format("%s-%s", this.getWorkerHostname(), this.getWorkerPort()); + this.workerId = String.format("%s-%s", this.getWorkerHostname(), this.getWorkerPort() != null + ? this.getWorkerPort() : this.getWorkerPortTls()); } return this.workerId; } diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/Worker.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/Worker.java index 218d4f0218d5e..0ff6dbc0431ec 100644 --- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/Worker.java +++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/Worker.java @@ -56,7 +56,7 @@ protected void start() throws Exception { workerService.start(getAuthenticationService(), getAuthorizationService(), errorNotifier); server = new WorkerServer(workerService, getAuthenticationService()); server.start(); - log.info("/** Started worker server on port={} **/", this.workerConfig.getWorkerPort()); + log.info("/** Started worker server **/"); try { errorNotifier.waitForError(); 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 b9aa35274be2b..7d2d294f9007b 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 @@ -87,9 +87,12 @@ private void init() { } List connectors = new ArrayList<>(); - httpConnector = new ServerConnector(server); - httpConnector.setPort(this.workerConfig.getWorkerPort()); - connectors.add(httpConnector); + if (this.workerConfig.getWorkerPort() != null) { + log.info("Configuring http server on port={}", this.workerConfig.getWorkerPort()); + httpConnector = new ServerConnector(server); + httpConnector.setPort(this.workerConfig.getWorkerPort()); + connectors.add(httpConnector); + } List handlers = new ArrayList<>(4); handlers.add(newServletContextHandler("/admin", @@ -125,6 +128,7 @@ private void init() { server.setHandler(stats); if (this.workerConfig.getTlsEnabled()) { + log.info("Configuring https server on port={}", this.workerConfig.getWorkerPortTls()); try { SslContextFactory sslCtxFactory; if (workerConfig.isTlsEnabledWithKeyStore()) { diff --git a/site2/docs/functions-worker.md b/site2/docs/functions-worker.md index 85e636e8bfa1f..2a06e02c211a0 100644 --- a/site2/docs/functions-worker.md +++ b/site2/docs/functions-worker.md @@ -109,7 +109,7 @@ To run function-worker separately, you have to configure the following parameter - `workerId`: The type is string. It is unique across clusters, which is used to identify a worker machine. - `workerHostname`: The hostname of the worker machine. -- `workerPort`: The port that the worker server listens on. Keep it as default if you don't customize it. +- `workerPort`: The port that the worker server listens on. Keep it as default if you don't customize it. Set it to `null` to disable the plaintext port. - `workerPortTls`: The TLS port that the worker server listens on. Keep it as default if you don't customize it. #### Function package parameter