Support authorization for java proxy (#74)

Fixed #58 

## Motivation

At first, I thought I needed to define the authorization cache configuration class when obtaining authorization, for example, by implementing a class similar to `ConfigurationCacheService`. However, after my research, I found that the current the class `AuthenticationService` value receives class `ConfigurationCacheService` as an argument, and this class uses zookeeper as a caching service. java-proxy runs in the same place as the pulsar cluster and inherits the function_worker configuration, so it can also use zookeeper as a caching service.

For the authentication plugin, the custom logic is implemented in jetty's filter, so there is no need to update the

Author: tuteng

java-proxy/src/main/java/io/streamnative/function/mesh/proxy/FunctionMeshProxyService.java

@@ -64,6 +64,9 @@ public class FunctionMeshProxyService implements WorkerService {
     private CustomObjectsApi customObjectsApi;
     private ApiClient apiClient;
 
+    private AuthenticationService authenticationService;
+    private AuthorizationService authorizationService;
+
     public FunctionMeshProxyService() {
 
     }
@@ -102,7 +105,8 @@ private void initKubernetesClient() throws IOException {
     public void start(AuthenticationService authenticationService,
                       AuthorizationService authorizationService,
                       ErrorNotifier errorNotifier) {
-        // https://github.com/streamnative/function-mesh/issues/58
+        this.authenticationService = authenticationService;
+        this.authorizationService = authorizationService;
     }
 
     public void stop() {

java-proxy/src/main/java/io/streamnative/function/mesh/proxy/FunctionMeshProxyWorker.java

@@ -18,24 +18,40 @@
  */
 package io.streamnative.function.mesh.proxy;
 
+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;
 import org.apache.pulsar.common.configuration.PulsarConfigurationLoader;
 import org.apache.pulsar.functions.worker.ErrorNotifier;
 import org.apache.pulsar.functions.worker.Worker;
 import org.apache.pulsar.functions.worker.WorkerConfig;
 import org.apache.pulsar.functions.worker.WorkerService;
 import org.apache.pulsar.functions.worker.rest.WorkerServer;
+import org.apache.pulsar.zookeeper.GlobalZooKeeperCache;
+import org.apache.pulsar.zookeeper.ZooKeeperClientFactory;
+import org.apache.pulsar.zookeeper.ZookeeperBkClientFactoryImpl;
+
+import java.io.IOException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 
 /**
  * This class for test.
  */
 @Slf4j
 public class FunctionMeshProxyWorker {
 
+    private ZooKeeperClientFactory zkClientFactory = null;
+    private final OrderedExecutor orderedExecutor = OrderedExecutor.newBuilder().numThreads(8).name("zk-cache-ordered").build();
+    private final ScheduledExecutorService cacheExecutor = Executors.newScheduledThreadPool(10,
+            new DefaultThreadFactory("zk-cache-callback"));
+    private GlobalZooKeeperCache globalZkCache;
+    private ConfigurationCacheService configurationCacheService;
     private final WorkerConfig workerConfig;
     private final WorkerService workerService;
     private final ErrorNotifier errorNotifier;
@@ -50,8 +66,7 @@ public FunctionMeshProxyWorker(WorkerConfig workerConfig) {
 
     protected void start() throws Exception {
         workerService.initAsStandalone(workerConfig);
-        // To do add authorization and authentication
-        workerService.start(getAuthenticationService(), null, errorNotifier);
+        workerService.start(getAuthenticationService(), getAuthorizationService(), errorNotifier);
         server = new WorkerServer(workerService, getAuthenticationService());
         server.start();
         log.info("/** Started worker server on port={} **/", this.workerConfig.getWorkerPort());
@@ -64,6 +79,37 @@ protected void start() throws Exception {
         }
     }
 
+    public ZooKeeperClientFactory getZooKeeperClientFactory() {
+        if (zkClientFactory == null) {
+            zkClientFactory = new ZookeeperBkClientFactoryImpl(orderedExecutor);
+        }
+        // Return default factory
+        return zkClientFactory;
+    }
+
+    private AuthorizationService getAuthorizationService() throws PulsarServerException {
+        if (this.workerConfig.isAuthorizationEnabled()) {
+            log.info("starting configuration cache service");
+
+            this.globalZkCache = new GlobalZooKeeperCache(getZooKeeperClientFactory(),
+                    (int) workerConfig.getZooKeeperSessionTimeoutMillis(),
+                    workerConfig.getZooKeeperOperationTimeoutSeconds(),
+                    workerConfig.getConfigurationStoreServers(),
+                    orderedExecutor, cacheExecutor,
+                    workerConfig.getZooKeeperOperationTimeoutSeconds());
+            try {
+                this.globalZkCache.start();
+            } catch (IOException e) {
+                throw new PulsarServerException(e);
+            }
+
+            this.configurationCacheService = new ConfigurationCacheService(
+                    this.globalZkCache, this.workerConfig.getPulsarFunctionsCluster());
+            return new AuthorizationService(getServiceConfiguration(), this.configurationCacheService);
+        }
+        return null;
+    }
+
     private AuthenticationService getAuthenticationService() throws PulsarServerException {
         return new AuthenticationService(getServiceConfiguration());
     }
@@ -79,5 +125,12 @@ protected void stop() {
             this.server.stop();
         }
         workerService.stop();
+        if (this.globalZkCache != null) {
+            try {
+                this.globalZkCache.close();
+            } catch (IOException e) {
+                log.warn("Failed to close global zk cache ", e);
+            }
+        }
     }
 }