convertFromConfig should support an overloading of cluster Name
@@ -22,6 +22,7 @@ import io.netty.util.concurrent.DefaultThreadFactory;
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 @@ public class Worker {
}
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 class Worker {
public Optional<Integer> getListenPortHTTPS() {
return this.server.getListenPortHTTPS();
}
+
+ private ServiceConfiguration getServiceConfiguration() {
+ ServiceConfiguration serviceConfiguration = PulsarConfigurationLoader.convertFrom(workerConfig);
+ serviceConfiguration.setClusterName(workerConfig.getPulsarFunctionsCluster());
+ return serviceConfiguration;
+ }
}
