separate function worker and broker client TLS configuration (apache#…

…6602) Fixes: apache#6521 ### Motivation Separate function worker and broker client TLS configuration
lujp · Apr 14, 2020 · 6f9eb5e · 6f9eb5e
1 parent 7cffe2a
commit 6f9eb5e
Show file tree

Hide file tree

Showing 3 changed files with 26 additions and 3 deletions.
diff --git a/conf/functions_worker.yml b/conf/functions_worker.yml
@@ -192,6 +192,12 @@ tlsAllowInsecureConnection: false
 # Tls cert refresh duration in seconds (set 0 to check on every new connection) 
 tlsCertRefreshCheckDurationSec: 300
 
+############################################
+# security settings for pulsar broker client
+############################################
+# The path to trusted certificates used by the Pulsar client to authenticate with Pulsar brokers
+brokerClientTrustCertsFilePath:
+
 ########################
 # State Management
 ########################

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
@@ -268,7 +268,7 @@ public class WorkerConfig implements Serializable, PulsarConfiguration {
     )
     private String tlsKeyFilePath;
     @FieldContext(
-        category = CATEGORY_SECURITY,
+        category = CATEGORY_WORKER_SECURITY,
         doc = "Path for the trusted TLS certificate file"
     )
     private String tlsTrustCertsFilePath = "";
@@ -333,6 +333,14 @@ public boolean getTlsEnabled() {
     	return tlsEnabled || workerPortTls != null;
     }
 
+    /******** security settings for pulsar broker client **********/
+
+    @FieldContext(
+            category = CATEGORY_CLIENT_SECURITY,
+            doc = "The path to trusted certificates used by the Pulsar client to authenticate with Pulsar brokers"
+    )
+    private String brokerClientTrustCertsFilePath;
+
 
     /******** Function Runtime configurations **********/
 

diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerService.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerService.java
@@ -125,9 +125,18 @@ public void start(URI dlogUri,
                     : workerConfig.getWorkerWebAddress();
 
             if (workerConfig.isAuthenticationEnabled()) {
+                // for compatible, if user do not define brokerClientTrustCertsFilePath, we will use tlsTrustCertsFilePath,
+                // otherwise we will use brokerClientTrustCertsFilePath
+                final String pulsarClientTlsTrustCertsFilePath;
+                if (StringUtils.isNotBlank(workerConfig.getBrokerClientTrustCertsFilePath())) {
+                    pulsarClientTlsTrustCertsFilePath = workerConfig.getBrokerClientTrustCertsFilePath();
+                } else {
+                    pulsarClientTlsTrustCertsFilePath = workerConfig.getTlsTrustCertsFilePath();
+                }
+
                 this.brokerAdmin = WorkerUtils.getPulsarAdminClient(workerConfig.getPulsarWebServiceUrl(),
                     workerConfig.getClientAuthenticationPlugin(), workerConfig.getClientAuthenticationParameters(),
-                    workerConfig.getTlsTrustCertsFilePath(), workerConfig.isTlsAllowInsecureConnection(),
+                    pulsarClientTlsTrustCertsFilePath, workerConfig.isTlsAllowInsecureConnection(),
                     workerConfig.isTlsHostnameVerificationEnable());
 
                 this.functionAdmin = WorkerUtils.getPulsarAdminClient(functionWebServiceUrl,
@@ -138,7 +147,7 @@ public void start(URI dlogUri,
                 this.client = WorkerUtils.getPulsarClient(this.workerConfig.getPulsarServiceUrl(),
                         workerConfig.getClientAuthenticationPlugin(),
                         workerConfig.getClientAuthenticationParameters(),
-                        workerConfig.isUseTls(), workerConfig.getTlsTrustCertsFilePath(),
+                        workerConfig.isUseTls(), pulsarClientTlsTrustCertsFilePath,
                         workerConfig.isTlsAllowInsecureConnection(), workerConfig.isTlsHostnameVerificationEnable());
             } else {
                 this.brokerAdmin = WorkerUtils.getPulsarAdminClient(workerConfig.getPulsarWebServiceUrl());