Get function cluster from broker config when start function worker wi…

…th broker (#10552)

### Motivation
When start function worker with broker, we need to set pulsarFunctionsCluster in functions_worker.yml, other wise it will fail to start.   #2328
If we run broker in k8s, we should set correct config map to change the default value in functions_worker.yml.
In this mode, pulsarFunctionsCluster should always be same with the cluster name in broker.conf, so we can get the setting directly from broker.conf.

### Modifications

get function cluster from broker config when start function worker with broker

pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java

@@ -1531,6 +1531,7 @@ public static WorkerConfig initializeWorkerConfigFromBrokerConfig(ServiceConfigu
         String hostname = ServiceConfigurationUtils.getDefaultOrConfiguredAddress(
                 ServiceConfigurationUtils.getAppliedAdvertisedAddress(brokerConfig));
         workerConfig.setWorkerHostname(hostname);
+        workerConfig.setPulsarFunctionsCluster(brokerConfig.getClusterName());
         // inherit broker authorization setting
         workerConfig.setAuthenticationEnabled(brokerConfig.isAuthenticationEnabled());
         workerConfig.setAuthenticationProviders(brokerConfig.getAuthenticationProviders());

pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionTlsTest.java

@@ -19,7 +19,6 @@
 package org.apache.pulsar.functions.worker;
 
 import static org.testng.Assert.assertEquals;
-
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.Sets;
@@ -121,7 +120,6 @@ void setup() throws Exception {
             tempDirectories[i] = PulsarFunctionTestTemporaryDirectory.create(getClass().getSimpleName());
             tempDirectories[i].useTemporaryDirectoriesForWorkerConfig(workerConfig);
             workerConfig.setPulsarFunctionsNamespace("public/functions");
-            workerConfig.setPulsarFunctionsCluster("my-cluster");
             workerConfig.setSchedulerClassName(
                 org.apache.pulsar.functions.worker.scheduler.RoundRobinScheduler.class.getName());
             workerConfig.setFunctionRuntimeFactoryClassName(ThreadRuntimeFactory.class.getName());

site2/docs/functions-worker.md

@@ -34,7 +34,6 @@ In this mode, most of the settings are already inherited from your broker config
 Pay attention to the following required settings when configuring functions-worker in this mode.
 
 - `numFunctionPackageReplicas`: The number of replicas to store function packages. The default value is `1`, which is good for standalone deployment. For production deployment, to ensure high availability, set it to be larger than `2`.
-- `pulsarFunctionsCluster`: Set the value to your Pulsar cluster name (same as the `clusterName` setting in the broker configuration).
 - `initializedDlogMetadata`: Whether to initialize distributed log metadata in runtime. If it is set to `true`, you must ensure that it has been initialized by `bin/pulsarinitialize-cluster-metadata` command.
 
 If authentication is enabled on the BookKeeper cluster, configure the following BookKeeper authentication settings.