[pulsar-function] support bookie authentication from function-worker (a…

…pache#4088)

### Motivation
Function worker uses bookie and zookeeper to store function's executable. However, function doesn't support authentication while connecting to bookie so, dbLog initialize fails when bookie requires auth&auth.

### Modification
Add bk-client authentication param to worker-config.

conf/functions_worker.yml

@@ -56,6 +56,16 @@ pulsarWebServiceUrl: http://localhost:8080
 # the authentication parameter to be used by the pulsar client used in worker service
 # clientAuthenticationParameters:
 
+# Bookie Authentication
+#
+# Authentication plugin to use when connecting to bookies
+# bookkeeperClientAuthenticationPlugin:
+# BookKeeper auth plugin implementatation specifics parameters name and values
+# bookkeeperClientAuthenticationParametersName:
+# Parameters for bookkeeper auth plugin
+# bookkeeperClientAuthenticationParameters:
+
+
 # pulsar topics used for function metadata management
 
 pulsarFunctionsNamespace: public/functions

deployment/terraform-ansible/templates/functions_worker.yml

@@ -36,6 +36,15 @@ downloadDirectory: /tmp/pulsar_functions
 processContainerFactory:
   logDirectory:
 
+# Bookie Authentication
+#
+# Authentication plugin to use when connecting to bookies
+# bookkeeperClientAuthenticationPlugin:
+# BookKeeper auth plugin implementatation specifics parameters name and values
+# bookkeeperClientAuthenticationParametersName:
+# Parameters for bookkeeper auth plugin
+# bookkeeperClientAuthenticationParameters:
+
 schedulerClassName: "org.apache.pulsar.functions.worker.scheduler.RoundRobinScheduler"
 functionAssignmentTopicName: "assignments"
 failureCheckFreqMs: 30000

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

@@ -222,6 +222,21 @@ public class WorkerConfig implements Serializable, PulsarConfiguration {
         doc = "The parameters of the authentication plugin used by function workers to talk to brokers"
     )
     private String clientAuthenticationParameters;
+    @FieldContext(
+        category = CATEGORY_CLIENT_SECURITY,
+        doc = "Authentication plugin to use when connecting to bookies"
+    )
+    private String bookkeeperClientAuthenticationPlugin;
+    @FieldContext(
+        category = CATEGORY_CLIENT_SECURITY,
+        doc = "BookKeeper auth plugin implementatation specifics parameters name and values"
+    )
+    private String bookkeeperClientAuthenticationParametersName;
+    @FieldContext(
+        category = CATEGORY_CLIENT_SECURITY,
+        doc = "Parameters for bookkeeper auth plugin"
+    )
+    private String bookkeeperClientAuthenticationParameters;
     @FieldContext(
         category = CATEGORY_FUNC_METADATA_MNG,
         doc = "Frequency how often worker performs compaction on function-topics, in seconds"

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

@@ -137,6 +137,14 @@ public static DistributedLogConfiguration getDlogConf(WorkerConfig workerConfig)
                 .setUseDaemonThread(true);
         conf.setProperty("bkc.allowShadedLedgerManagerFactoryClass", true);
         conf.setProperty("bkc.shadedLedgerManagerFactoryClassPrefix", "dlshade.");
+        if (isNotBlank(workerConfig.getBookkeeperClientAuthenticationPlugin())) {
+            conf.setProperty("bkc.clientAuthProviderFactoryClass",
+                    workerConfig.getBookkeeperClientAuthenticationPlugin());
+            if (isNotBlank(workerConfig.getBookkeeperClientAuthenticationParametersName())) {
+                conf.setProperty("bkc." + workerConfig.getBookkeeperClientAuthenticationParametersName(),
+                        workerConfig.getBookkeeperClientAuthenticationParameters());
+            }
+        }
         return conf;
     }