Merge pull request #12 from freeznet/fix_forward_request_to_leader

sijie · Feb 4, 2021 · cb74f28 · cb74f28
2 parents b3a2d35 + de0060f
commit cb74f28
Show file tree

Hide file tree

Showing 4 changed files with 14 additions and 7 deletions.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
@@ -1388,11 +1388,6 @@ public static WorkerConfig initializeWorkerConfigFromBrokerConfig(ServiceConfigu
         String hostname = ServiceConfigurationUtils.getDefaultOrConfiguredAddress(
             brokerConfig.getAdvertisedAddress());
         workerConfig.setWorkerHostname(hostname);
-        workerConfig.setWorkerId(
-            "c-" + brokerConfig.getClusterName()
-                + "-fw-" + hostname
-                + "-" + (workerConfig.getTlsEnabled()
-                    ? workerConfig.getWorkerPortTls() : workerConfig.getWorkerPort()));
         // inherit broker authorization setting
         workerConfig.setAuthenticationEnabled(brokerConfig.isAuthenticationEnabled());
         workerConfig.setAuthenticationProviders(brokerConfig.getAuthenticationProviders());
@@ -1403,6 +1398,7 @@ public static WorkerConfig initializeWorkerConfigFromBrokerConfig(ServiceConfigu
         workerConfig.setZooKeeperOperationTimeoutSeconds(brokerConfig.getZooKeeperOperationTimeoutSeconds());
 
         workerConfig.setTlsAllowInsecureConnection(brokerConfig.isTlsAllowInsecureConnection());
+        workerConfig.setTlsEnabled(brokerConfig.isTlsEnabled());
         workerConfig.setTlsEnableHostnameVerification(false);
         workerConfig.setBrokerClientTrustCertsFilePath(brokerConfig.getTlsTrustCertsFilePath());
 
@@ -1418,6 +1414,12 @@ public static WorkerConfig initializeWorkerConfigFromBrokerConfig(ServiceConfigu
             workerConfig.setFunctionsWorkerServiceNarPackage(
                 brokerConfig.getFunctionsWorkerServiceNarPackage());
         }
+
+        workerConfig.setWorkerId(
+                "c-" + brokerConfig.getClusterName()
+                        + "-fw-" + hostname
+                        + "-" + (workerConfig.getTlsEnabled()
+                        ? workerConfig.getWorkerPortTls() : workerConfig.getWorkerPort()));
         return workerConfig;
     }
 }
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionTlsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionTlsTest.java
@@ -111,6 +111,7 @@ void setup() throws Exception {
             config.setBrokerClientAuthenticationParameters(
                 "tlsCertFile:" + TLS_CLIENT_CERT_FILE_PATH + ",tlsKeyFile:" + TLS_CLIENT_KEY_FILE_PATH);
             config.setFunctionsWorkerEnabled(true);
+            config.setTlsEnabled(true);
 
             WorkerConfig workerConfig = PulsarService.initializeWorkerConfigFromBrokerConfig(config, null);
             workerConfig.setPulsarFunctionsNamespace("public/functions");
@@ -190,6 +191,8 @@ public void testFunctionsCreation() throws Exception {
             FunctionConfig functionConfig = createFunctionConfig(jarFilePathUrl, testTenant, "my-ns",
                 functionName, "my.*", "sink-topic-" + i, "sub-" + i);
 
+            log.info(" -------- Start test function : {}", functionName);
+
             pulsarAdmins[i].functions().createFunctionWithUrl(
                 functionConfig, jarFilePathUrl
             );
@@ -198,6 +201,8 @@ public void testFunctionsCreation() throws Exception {
             assertEquals(config.getTenant(), testTenant);
             assertEquals(config.getNamespace(), "my-ns");
             assertEquals(config.getName(), functionName);
+
+            pulsarAdmins[i].functions().deleteFunction(config.getTenant(), config.getNamespace(), config.getName());
         }
     }
 

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
@@ -397,7 +397,7 @@ public boolean isBrokerClientAuthenticationEnabled() {
     private Properties properties = new Properties();
 
     public boolean getTlsEnabled() {
-    	return tlsEnabled || workerPortTls != null;
+    	return tlsEnabled && workerPortTls != null;
     }
 
     @FieldContext(

diff --git a/...-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/WorkerServer.java b/...-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/WorkerServer.java
@@ -137,7 +137,7 @@ private void init() {
         handlers.add(stats);
         server.setHandler(stats);
 
-        if (this.workerConfig.getWorkerPortTls() != null) {
+        if (this.workerConfig.getTlsEnabled()) {
             try {
                 SslContextFactory sslCtxFactory = SecurityUtility.createSslContextFactory(
                         this.workerConfig.isTlsAllowInsecureConnection(), this.workerConfig.getTlsTrustCertsFilePath(),