convertFromConfig should support an overloading of cluster Name (apac…

…he#7890)

Co-authored-by: Sanjeev Kulkarni <[email protected]>

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

@@ -22,6 +22,7 @@
 import lombok.extern.slf4j.Slf4j;
 import org.apache.bookkeeper.common.util.OrderedExecutor;
 import org.apache.pulsar.broker.PulsarServerException;
+import org.apache.pulsar.broker.ServiceConfiguration;
 import org.apache.pulsar.broker.authentication.AuthenticationService;
 import org.apache.pulsar.broker.authorization.AuthorizationService;
 import org.apache.pulsar.broker.cache.ConfigurationCacheService;
@@ -179,13 +180,13 @@ private AuthorizationService getAuthorizationService() throws PulsarServerExcept
             }
 
             this.configurationCacheService = new ConfigurationCacheService(this.globalZkCache, this.workerConfig.getPulsarFunctionsCluster());
-                return new AuthorizationService(PulsarConfigurationLoader.convertFrom(workerConfig), this.configurationCacheService);
+                return new AuthorizationService(getServiceConfiguration(), this.configurationCacheService);
             }
         return null;
     }
 
     private AuthenticationService getAuthenticationService() throws PulsarServerException {
-        return new AuthenticationService(PulsarConfigurationLoader.convertFrom(workerConfig));
+        return new AuthenticationService(getServiceConfiguration());
     }
 
     public ZooKeeperClientFactory getZooKeeperClientFactory() {
@@ -223,4 +224,10 @@ public Optional<Integer> getListenPortHTTP() {
     public Optional<Integer> getListenPortHTTPS() {
         return this.server.getListenPortHTTPS();
     }
+
+    private ServiceConfiguration getServiceConfiguration() {
+        ServiceConfiguration serviceConfiguration = PulsarConfigurationLoader.convertFrom(workerConfig);
+        serviceConfiguration.setClusterName(workerConfig.getPulsarFunctionsCluster());
+        return serviceConfiguration;
+    }
 }