You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2020/07/03 06:36:25 UTC
[pulsar] branch master updated: [Issue: 7379] Improve security
setting of Pulsar Functions (#7424)
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new e99b669 [Issue: 7379] Improve security setting of Pulsar Functions (#7424)
e99b669 is described below
commit e99b66933e4775efbb7c7df8ea3a38047d879d6b
Author: 冉小龙 <rx...@apache.org>
AuthorDate: Fri Jul 3 14:36:11 2020 +0800
[Issue: 7379] Improve security setting of Pulsar Functions (#7424)
Fixes #7379
### Motivation
Rename some settings to make those broker-client TLS settings clearer.
### Modifications
- replace `clientAuthenticationParameters` with `brokerClientAuthenticationParameters`
- replace `clientAuthenticationPlugin` with `brokerClientAuthenticationPlugin`
- replace `tlsHostnameVerificationEnable` with `tlsEnableHostnameVerification`
---
conf/functions_worker.yml | 26 ++++++++++------
.../org/apache/pulsar/PulsarBrokerStarter.java | 9 +++---
.../java/org/apache/pulsar/PulsarStandalone.java | 9 +++---
.../worker/PulsarFunctionE2ESecurityTest.java | 12 ++++----
.../worker/PulsarFunctionLocalRunTest.java | 12 ++++----
.../worker/PulsarFunctionPublishTest.java | 12 ++++----
.../apache/pulsar/io/PulsarFunctionAdminTest.java | 12 ++++----
.../apache/pulsar/io/PulsarFunctionE2ETest.java | 12 ++++----
.../apache/pulsar/io/PulsarFunctionTlsTest.java | 4 +--
.../pulsar/functions/worker/WorkerConfig.java | 35 ++++++++++++++++++++--
.../functions/worker/FunctionRuntimeManager.java | 6 ++--
.../org/apache/pulsar/functions/worker/Worker.java | 4 +--
.../pulsar/functions/worker/WorkerService.java | 14 ++++-----
13 files changed, 101 insertions(+), 66 deletions(-)
diff --git a/conf/functions_worker.yml b/conf/functions_worker.yml
index 4ecf1e0..7ab160e 100644
--- a/conf/functions_worker.yml
+++ b/conf/functions_worker.yml
@@ -48,13 +48,25 @@ downloadDirectory: download/pulsar_functions
# Configure the pulsar client used by function metadata management
#
-# points
+# points
+# Whether to enable TLS when clients connect to broker
+useTls: false
+# For TLS:
+# brokerServiceUrl=pulsar+ssl://localhost:6651/
pulsarServiceUrl: pulsar://localhost:6650
+# For TLS:
+# webServiceUrl=https://localhost:8443/
pulsarWebServiceUrl: http://localhost:8080
+
+############################################
+# security settings for pulsar broker client
+############################################
+# The path to trusted certificates used by the Pulsar client to authenticate with Pulsar brokers
+# brokerClientTrustCertsFilePath:
# the authentication plugin to be used by the pulsar client used in worker service
-# clientAuthenticationPlugin:
+# brokerClientAuthenticationPlugin:
# the authentication parameter to be used by the pulsar client used in worker service
-# clientAuthenticationParameters:
+# brokerClientAuthenticationParameters:
# Bookie Authentication
#
@@ -193,15 +205,11 @@ tlsKeyFilePath:
tlsTrustCertsFilePath:
# Accept untrusted TLS certificate from client
tlsAllowInsecureConnection: false
+# Whether server hostname must match the common name of the certificate
+tlsEnableHostnameVerification: 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-broker/src/main/java/org/apache/pulsar/PulsarBrokerStarter.java b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarBrokerStarter.java
index e988df1..165b616 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarBrokerStarter.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarBrokerStarter.java
@@ -189,14 +189,13 @@ public class PulsarBrokerStarter {
workerConfig.setZooKeeperSessionTimeoutMillis(brokerConfig.getZooKeeperSessionTimeoutMillis());
workerConfig.setZooKeeperOperationTimeoutSeconds(brokerConfig.getZooKeeperOperationTimeoutSeconds());
- workerConfig.setTlsHostnameVerificationEnable(false);
-
workerConfig.setTlsAllowInsecureConnection(brokerConfig.isTlsAllowInsecureConnection());
- workerConfig.setTlsTrustCertsFilePath(brokerConfig.getTlsTrustCertsFilePath());
+ workerConfig.setTlsEnableHostnameVerification(false);
+ workerConfig.setBrokerClientTrustCertsFilePath(brokerConfig.getTlsTrustCertsFilePath());
// client in worker will use this config to authenticate with broker
- workerConfig.setClientAuthenticationPlugin(brokerConfig.getBrokerClientAuthenticationPlugin());
- workerConfig.setClientAuthenticationParameters(brokerConfig.getBrokerClientAuthenticationParameters());
+ workerConfig.setBrokerClientAuthenticationPlugin(brokerConfig.getBrokerClientAuthenticationPlugin());
+ workerConfig.setBrokerClientAuthenticationParameters(brokerConfig.getBrokerClientAuthenticationParameters());
// inherit super users
workerConfig.setSuperUserRoles(brokerConfig.getSuperUserRoles());
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 3fab76f..cbe1450 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarStandalone.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarStandalone.java
@@ -299,14 +299,13 @@ public class PulsarStandalone implements AutoCloseable {
workerConfig.setZooKeeperSessionTimeoutMillis(config.getZooKeeperSessionTimeoutMillis());
workerConfig.setZooKeeperOperationTimeoutSeconds(config.getZooKeeperOperationTimeoutSeconds());
- workerConfig.setTlsHostnameVerificationEnable(false);
-
workerConfig.setTlsAllowInsecureConnection(config.isTlsAllowInsecureConnection());
- workerConfig.setTlsTrustCertsFilePath(config.getTlsTrustCertsFilePath());
+ workerConfig.setTlsEnableHostnameVerification(false);
+ workerConfig.setBrokerClientTrustCertsFilePath(config.getTlsTrustCertsFilePath());
// client in worker will use this config to authenticate with broker
- workerConfig.setClientAuthenticationPlugin(config.getBrokerClientAuthenticationPlugin());
- workerConfig.setClientAuthenticationParameters(config.getBrokerClientAuthenticationParameters());
+ workerConfig.setBrokerClientAuthenticationPlugin(config.getBrokerClientAuthenticationPlugin());
+ workerConfig.setBrokerClientAuthenticationParameters(config.getBrokerClientAuthenticationParameters());
// inherit super users
workerConfig.setSuperUserRoles(config.getSuperUserRoles());
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionE2ESecurityTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionE2ESecurityTest.java
index 03f9bd7..4f332e6 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionE2ESecurityTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionE2ESecurityTest.java
@@ -177,10 +177,10 @@ public class PulsarFunctionE2ESecurityTest {
ClientBuilder clientBuilder = PulsarClient.builder().serviceUrl(this.workerConfig.getPulsarServiceUrl())
.operationTimeout(1000, TimeUnit.MILLISECONDS);
- if (isNotBlank(workerConfig.getClientAuthenticationPlugin())
- && isNotBlank(workerConfig.getClientAuthenticationParameters())) {
- clientBuilder.authentication(workerConfig.getClientAuthenticationPlugin(),
- workerConfig.getClientAuthenticationParameters());
+ if (isNotBlank(workerConfig.getBrokerClientAuthenticationPlugin())
+ && isNotBlank(workerConfig.getBrokerClientAuthenticationParameters())) {
+ clientBuilder.authentication(workerConfig.getBrokerClientAuthenticationPlugin(),
+ workerConfig.getBrokerClientAuthenticationParameters());
}
pulsarClient = clientBuilder.build();
@@ -242,8 +242,8 @@ public class PulsarFunctionE2ESecurityTest {
workerConfig.setWorkerHostname(hostname);
workerConfig.setWorkerId(workerId);
- workerConfig.setClientAuthenticationPlugin(AuthenticationToken.class.getName());
- workerConfig.setClientAuthenticationParameters(
+ workerConfig.setBrokerClientAuthenticationPlugin(AuthenticationToken.class.getName());
+ workerConfig.setBrokerClientAuthenticationParameters(
String.format("token:%s", adminToken));
workerConfig.setAuthenticationEnabled(config.isAuthenticationEnabled());
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionLocalRunTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionLocalRunTest.java
index 0fe2958..3874dff 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionLocalRunTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionLocalRunTest.java
@@ -205,12 +205,12 @@ public class PulsarFunctionLocalRunTest {
ClientBuilder clientBuilder = PulsarClient.builder()
.serviceUrl(pulsar.getBrokerServiceUrl());
- if (isNotBlank(workerConfig.getClientAuthenticationPlugin())
- && isNotBlank(workerConfig.getClientAuthenticationParameters())) {
+ if (isNotBlank(workerConfig.getBrokerClientAuthenticationPlugin())
+ && isNotBlank(workerConfig.getBrokerClientAuthenticationParameters())) {
clientBuilder.enableTls(workerConfig.isUseTls());
clientBuilder.allowTlsInsecureConnection(workerConfig.isTlsAllowInsecureConnection());
- clientBuilder.authentication(workerConfig.getClientAuthenticationPlugin(),
- workerConfig.getClientAuthenticationParameters());
+ clientBuilder.authentication(workerConfig.getBrokerClientAuthenticationPlugin(),
+ workerConfig.getBrokerClientAuthenticationParameters());
clientBuilder.serviceUrl(pulsar.getBrokerServiceUrlTls());
}
pulsarClient = clientBuilder.build();
@@ -312,8 +312,8 @@ public class PulsarFunctionLocalRunTest {
workerConfig.setWorkerHostname(hostname);
workerConfig.setWorkerId(workerId);
- workerConfig.setClientAuthenticationPlugin(AuthenticationTls.class.getName());
- workerConfig.setClientAuthenticationParameters(
+ workerConfig.setBrokerClientAuthenticationPlugin(AuthenticationTls.class.getName());
+ workerConfig.setBrokerClientAuthenticationParameters(
String.format("tlsCertFile:%s,tlsKeyFile:%s", TLS_CLIENT_CERT_FILE_PATH, TLS_CLIENT_KEY_FILE_PATH));
workerConfig.setUseTls(true);
workerConfig.setTlsAllowInsecureConnection(true);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionPublishTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionPublishTest.java
index 90882bb..4ce7827 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionPublishTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionPublishTest.java
@@ -189,12 +189,12 @@ public class PulsarFunctionPublishTest {
admin.clusters().updateCluster(config.getClusterName(), clusterData);
ClientBuilder clientBuilder = PulsarClient.builder().serviceUrl(this.workerConfig.getPulsarServiceUrl());
- if (isNotBlank(workerConfig.getClientAuthenticationPlugin())
- && isNotBlank(workerConfig.getClientAuthenticationParameters())) {
+ if (isNotBlank(workerConfig.getBrokerClientAuthenticationPlugin())
+ && isNotBlank(workerConfig.getBrokerClientAuthenticationParameters())) {
clientBuilder.enableTls(workerConfig.isUseTls());
clientBuilder.allowTlsInsecureConnection(workerConfig.isTlsAllowInsecureConnection());
- clientBuilder.authentication(workerConfig.getClientAuthenticationPlugin(),
- workerConfig.getClientAuthenticationParameters());
+ clientBuilder.authentication(workerConfig.getBrokerClientAuthenticationPlugin(),
+ workerConfig.getBrokerClientAuthenticationParameters());
}
pulsarClient = clientBuilder.build();
@@ -244,8 +244,8 @@ public class PulsarFunctionPublishTest {
workerConfig.setWorkerHostname(hostname);
workerConfig.setWorkerId(workerId);
- workerConfig.setClientAuthenticationPlugin(AuthenticationTls.class.getName());
- workerConfig.setClientAuthenticationParameters(
+ workerConfig.setBrokerClientAuthenticationPlugin(AuthenticationTls.class.getName());
+ workerConfig.setBrokerClientAuthenticationParameters(
String.format("tlsCertFile:%s,tlsKeyFile:%s", TLS_CLIENT_CERT_FILE_PATH, TLS_CLIENT_KEY_FILE_PATH));
workerConfig.setUseTls(true);
workerConfig.setTlsAllowInsecureConnection(true);
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 28fd0b4..3b96636 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
@@ -142,12 +142,12 @@ public class PulsarFunctionAdminTest {
admin.clusters().updateCluster(config.getClusterName(), clusterData);
ClientBuilder clientBuilder = PulsarClient.builder().serviceUrl(this.workerConfig.getPulsarServiceUrl());
- if (isNotBlank(workerConfig.getClientAuthenticationPlugin())
- && isNotBlank(workerConfig.getClientAuthenticationParameters())) {
+ if (isNotBlank(workerConfig.getBrokerClientAuthenticationPlugin())
+ && isNotBlank(workerConfig.getBrokerClientAuthenticationParameters())) {
clientBuilder.enableTls(workerConfig.isUseTls());
clientBuilder.allowTlsInsecureConnection(workerConfig.isTlsAllowInsecureConnection());
- clientBuilder.authentication(workerConfig.getClientAuthenticationPlugin(),
- workerConfig.getClientAuthenticationParameters());
+ clientBuilder.authentication(workerConfig.getBrokerClientAuthenticationPlugin(),
+ workerConfig.getBrokerClientAuthenticationParameters());
}
pulsarClient = clientBuilder.build();
@@ -192,8 +192,8 @@ public class PulsarFunctionAdminTest {
workerConfig
.setWorkerId("c-" + config.getClusterName() + "-fw-" + hostname + "-" + workerConfig.getWorkerPort());
- workerConfig.setClientAuthenticationPlugin(AuthenticationTls.class.getName());
- workerConfig.setClientAuthenticationParameters(
+ workerConfig.setBrokerClientAuthenticationPlugin(AuthenticationTls.class.getName());
+ workerConfig.setBrokerClientAuthenticationParameters(
String.format("tlsCertFile:%s,tlsKeyFile:%s", TLS_CLIENT_CERT_FILE_PATH, TLS_CLIENT_KEY_FILE_PATH));
workerConfig.setUseTls(true);
workerConfig.setTlsAllowInsecureConnection(true);
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 f953ca4..e9e3b68 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
@@ -219,12 +219,12 @@ public class PulsarFunctionE2ETest {
admin.clusters().updateCluster(config.getClusterName(), clusterData);
ClientBuilder clientBuilder = PulsarClient.builder().serviceUrl(this.workerConfig.getPulsarServiceUrl());
- if (isNotBlank(workerConfig.getClientAuthenticationPlugin())
- && isNotBlank(workerConfig.getClientAuthenticationParameters())) {
+ if (isNotBlank(workerConfig.getBrokerClientAuthenticationPlugin())
+ && isNotBlank(workerConfig.getBrokerClientAuthenticationParameters())) {
clientBuilder.enableTls(workerConfig.isUseTls());
clientBuilder.allowTlsInsecureConnection(workerConfig.isTlsAllowInsecureConnection());
- clientBuilder.authentication(workerConfig.getClientAuthenticationPlugin(),
- workerConfig.getClientAuthenticationParameters());
+ clientBuilder.authentication(workerConfig.getBrokerClientAuthenticationPlugin(),
+ workerConfig.getBrokerClientAuthenticationParameters());
}
pulsarClient = clientBuilder.build();
@@ -332,8 +332,8 @@ public class PulsarFunctionE2ETest {
workerConfig.setWorkerHostname(hostname);
workerConfig.setWorkerId(workerId);
- workerConfig.setClientAuthenticationPlugin(AuthenticationTls.class.getName());
- workerConfig.setClientAuthenticationParameters(
+ workerConfig.setBrokerClientAuthenticationPlugin(AuthenticationTls.class.getName());
+ workerConfig.setBrokerClientAuthenticationParameters(
String.format("tlsCertFile:%s,tlsKeyFile:%s", TLS_CLIENT_CERT_FILE_PATH, TLS_CLIENT_KEY_FILE_PATH));
workerConfig.setUseTls(true);
workerConfig.setTlsAllowInsecureConnection(true);
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 e241f76..7fa1608 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
@@ -192,8 +192,8 @@ public class PulsarFunctionTlsTest {
workerConfig.setWorkerHostname(hostname);
workerConfig.setWorkerId(workerId);
- workerConfig.setClientAuthenticationPlugin(AuthenticationTls.class.getName());
- workerConfig.setClientAuthenticationParameters(
+ workerConfig.setBrokerClientAuthenticationPlugin(AuthenticationTls.class.getName());
+ workerConfig.setBrokerClientAuthenticationParameters(
String.format("tlsCertFile:%s,tlsKeyFile:%s", TLS_CLIENT_CERT_FILE_PATH, TLS_CLIENT_KEY_FILE_PATH));
workerConfig.setUseTls(true);
workerConfig.setTlsAllowInsecureConnection(true);
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 021c12a..4b0ddd1 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
@@ -260,12 +260,12 @@ public class WorkerConfig implements Serializable, PulsarConfiguration {
category = CATEGORY_CLIENT_SECURITY,
doc = "The authentication plugin used by function workers to talk to brokers"
)
- private String clientAuthenticationPlugin;
+ private String brokerClientAuthenticationPlugin;
@FieldContext(
category = CATEGORY_CLIENT_SECURITY,
doc = "The parameters of the authentication plugin used by function workers to talk to brokers"
)
- private String clientAuthenticationParameters;
+ private String brokerClientAuthenticationParameters;
@FieldContext(
category = CATEGORY_CLIENT_SECURITY,
doc = "Authentication plugin to use when connecting to bookies"
@@ -331,7 +331,7 @@ public class WorkerConfig implements Serializable, PulsarConfiguration {
category = CATEGORY_SECURITY,
doc = "Whether to enable hostname verification on TLS connections"
)
- private boolean tlsHostnameVerificationEnable = false;
+ private boolean tlsEnableHostnameVerification = false;
@FieldContext(
category = CATEGORY_SECURITY,
doc = "Tls cert refresh duration in seconds (set 0 to check on every new connection)"
@@ -561,4 +561,33 @@ public class WorkerConfig implements Serializable, PulsarConfiguration {
)
@Deprecated
private KubernetesContainerFactory kubernetesContainerFactory;
+
+ @FieldContext(
+ category = CATEGORY_CLIENT_SECURITY,
+ doc = "The parameters of the authentication plugin used by function workers to talk to brokers"
+ )
+ @Deprecated
+ private String clientAuthenticationParameters;
+ @FieldContext(
+ category = CATEGORY_CLIENT_SECURITY,
+ doc = "The authentication plugin used by function workers to talk to brokers"
+ )
+ @Deprecated
+ private String clientAuthenticationPlugin;
+
+ public String getBrokerClientAuthenticationPlugin() {
+ if (null == brokerClientAuthenticationPlugin) {
+ return clientAuthenticationPlugin;
+ } else {
+ return brokerClientAuthenticationPlugin;
+ }
+ }
+
+ public String getBrokerClientAuthenticationParameters() {
+ if (null == brokerClientAuthenticationParameters) {
+ return clientAuthenticationParameters;
+ } else {
+ return brokerClientAuthenticationParameters;
+ }
+ }
}
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java
index c8fb8fd..3eba9da 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java
@@ -164,12 +164,12 @@ public class FunctionRuntimeManager implements AutoCloseable{
AuthenticationConfig authConfig = null;
if (workerConfig.isAuthenticationEnabled()) {
authConfig = AuthenticationConfig.builder()
- .clientAuthenticationPlugin(workerConfig.getClientAuthenticationPlugin())
- .clientAuthenticationParameters(workerConfig.getClientAuthenticationParameters())
+ .clientAuthenticationPlugin(workerConfig.getBrokerClientAuthenticationPlugin())
+ .clientAuthenticationParameters(workerConfig.getBrokerClientAuthenticationParameters())
.tlsTrustCertsFilePath(workerConfig.getTlsTrustCertsFilePath())
.useTls(workerConfig.isUseTls())
.tlsAllowInsecureConnection(workerConfig.isTlsAllowInsecureConnection())
- .tlsHostnameVerificationEnable(workerConfig.isTlsHostnameVerificationEnable())
+ .tlsHostnameVerificationEnable(workerConfig.isTlsEnableHostnameVerification())
.build();
//initialize function authentication provider
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 ccd3d3a..0b9eb77 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
@@ -85,9 +85,9 @@ public class Worker {
throws InterruptedException, PulsarAdminException, IOException {
// initializing pulsar functions namespace
PulsarAdmin admin = WorkerUtils.getPulsarAdminClient(workerConfig.getPulsarWebServiceUrl(),
- workerConfig.getClientAuthenticationPlugin(), workerConfig.getClientAuthenticationParameters(),
+ workerConfig.getBrokerClientAuthenticationPlugin(), workerConfig.getBrokerClientAuthenticationParameters(),
workerConfig.getTlsTrustCertsFilePath(), workerConfig.isTlsAllowInsecureConnection(),
- workerConfig.isTlsHostnameVerificationEnable());
+ workerConfig.isTlsEnableHostnameVerification());
InternalConfigurationData internalConf;
// make sure pulsar broker is up
log.info("Checking if pulsar service at {} is up...", workerConfig.getPulsarWebServiceUrl());
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
index 35dff9e..7754a82 100644
--- 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
@@ -137,20 +137,20 @@ public class WorkerService {
}
this.brokerAdmin = WorkerUtils.getPulsarAdminClient(workerConfig.getPulsarWebServiceUrl(),
- workerConfig.getClientAuthenticationPlugin(), workerConfig.getClientAuthenticationParameters(),
+ workerConfig.getBrokerClientAuthenticationPlugin(), workerConfig.getBrokerClientAuthenticationParameters(),
pulsarClientTlsTrustCertsFilePath, workerConfig.isTlsAllowInsecureConnection(),
- workerConfig.isTlsHostnameVerificationEnable());
+ workerConfig.isTlsEnableHostnameVerification());
this.functionAdmin = WorkerUtils.getPulsarAdminClient(functionWebServiceUrl,
- workerConfig.getClientAuthenticationPlugin(), workerConfig.getClientAuthenticationParameters(),
+ workerConfig.getBrokerClientAuthenticationPlugin(), workerConfig.getBrokerClientAuthenticationParameters(),
workerConfig.getTlsTrustCertsFilePath(), workerConfig.isTlsAllowInsecureConnection(),
- workerConfig.isTlsHostnameVerificationEnable());
+ workerConfig.isTlsEnableHostnameVerification());
this.client = WorkerUtils.getPulsarClient(workerConfig.getPulsarServiceUrl(),
- workerConfig.getClientAuthenticationPlugin(),
- workerConfig.getClientAuthenticationParameters(),
+ workerConfig.getBrokerClientAuthenticationPlugin(),
+ workerConfig.getBrokerClientAuthenticationParameters(),
workerConfig.isUseTls(), pulsarClientTlsTrustCertsFilePath,
- workerConfig.isTlsAllowInsecureConnection(), workerConfig.isTlsHostnameVerificationEnable());
+ workerConfig.isTlsAllowInsecureConnection(), workerConfig.isTlsEnableHostnameVerification());
} else {
this.brokerAdmin = WorkerUtils.getPulsarAdminClient(workerConfig.getPulsarWebServiceUrl());