Splitting the authentication logic of function worker and client (apa…

…che#8824)

Fixes apache#8338 

### Motivation

>In some scenarios, users use their own function-worker to connect to an existing pulsar cluster. Their own function-worker and pulsar cluster have different authentication methods, In the following code, when both function-worker and client have enabled the authentication and authorization services, the authentication and authorization can take effect. A better way is to separate them. function-worker can enable and disable the authentication service, and the broker-client can also enable and disable the authentication service according to the configuration.

### Modifications

Add a configuration called `brokerClientAuthenticationEnabled` in the configuration file, which is disabled by default. It is used to control whether the broker-client of function-worker enable or disable the authentication.

conf/functions_worker.yml

@@ -61,6 +61,7 @@ pulsarWebServiceUrl: http://localhost:8080
 ############################################
 # security settings for pulsar broker client
 ############################################
+brokerClientAuthenticationEnabled: false
 # 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

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java

@@ -273,6 +273,18 @@ public class WorkerConfig implements Serializable, PulsarConfiguration {
         doc = "The frequency of instance liveness check, in milliseconds"
     )
     private long instanceLivenessCheckFreqMs;
+    @FieldContext(
+            category = CATEGORY_CLIENT_SECURITY,
+            doc = "Whether to enable the broker client authentication used by function workers to talk to brokers"
+    )
+    private Boolean brokerClientAuthenticationEnabled = null;
+    public boolean isBrokerClientAuthenticationEnabled() {
+        if (brokerClientAuthenticationEnabled != null) {
+            return brokerClientAuthenticationEnabled;
+        } else {
+            return authenticationEnabled;
+        }
+    }
     @FieldContext(
         category = CATEGORY_CLIENT_SECURITY,
         doc = "The authentication plugin used by function workers to talk to brokers"

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerService.java

@@ -133,7 +133,8 @@ public void start(URI dlogUri,
                     ? workerConfig.getFunctionWebServiceUrl()
                     : workerConfig.getWorkerWebAddress();
 
-            if (workerConfig.isAuthenticationEnabled()) {
+             // using isBrokerClientAuthenticationEnabled instead of isAuthenticationEnabled in function-worker
+            if (workerConfig.isBrokerClientAuthenticationEnabled()) {
                 // for compatible, if user do not define brokerClientTrustCertsFilePath, we will use tlsTrustCertsFilePath,
                 // otherwise we will use brokerClientTrustCertsFilePath
                 final String pulsarClientTlsTrustCertsFilePath;
@@ -142,7 +143,6 @@ public void start(URI dlogUri,
                 } else {
                     pulsarClientTlsTrustCertsFilePath = workerConfig.getTlsTrustCertsFilePath();
                 }
-
                 this.brokerAdmin = WorkerUtils.getPulsarAdminClient(workerConfig.getPulsarWebServiceUrl(),
                     workerConfig.getBrokerClientAuthenticationPlugin(), workerConfig.getBrokerClientAuthenticationParameters(),
                     pulsarClientTlsTrustCertsFilePath, workerConfig.isTlsAllowInsecureConnection(),