Moved ClassLoading and Reflections Helper functions to common

pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionTlsTest.java

@@ -52,11 +52,11 @@
 import org.apache.pulsar.client.impl.auth.AuthenticationTls;
 import org.apache.pulsar.common.functions.FunctionConfig;
 import org.apache.pulsar.common.policies.data.TenantInfo;
+import org.apache.pulsar.common.util.ClassLoaderUtils;
 import org.apache.pulsar.common.util.ObjectMapperFactory;
 import org.apache.pulsar.functions.api.utils.IdentityFunction;
 import org.apache.pulsar.functions.runtime.thread.ThreadRuntimeFactory;
 import org.apache.pulsar.functions.sink.PulsarSink;
-import org.apache.pulsar.functions.utils.FunctionCommon;
 import org.apache.pulsar.functions.worker.FunctionMetaDataManager;
 import org.apache.pulsar.functions.runtime.thread.ThreadRuntimeFactoryConfig;
 import org.apache.pulsar.functions.worker.WorkerConfig;
@@ -237,7 +237,7 @@ protected static FunctionConfig createFunctionConfig(String jarFile, String tena
 
         File file = new File(jarFile);
         try {
-            FunctionCommon.loadJar(file);
+            ClassLoaderUtils.loadJar(file);
         } catch (MalformedURLException e) {
             throw new RuntimeException("Failed to load user jar " + file, e);
         }

pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/CmdFunctionsTest.java

@@ -57,7 +57,7 @@
 import org.apache.pulsar.functions.api.Context;
 import org.apache.pulsar.functions.api.Function;
 import org.apache.pulsar.functions.api.utils.IdentityFunction;
-import org.apache.pulsar.functions.utils.Reflections;
+import org.apache.pulsar.common.util.Reflections;
 import org.apache.pulsar.functions.utils.FunctionCommon;
 import org.powermock.core.classloader.annotations.PowerMockIgnore;
 import org.powermock.core.classloader.annotations.PrepareForTest;

pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdSinks.java

@@ -44,7 +44,7 @@
 import org.apache.pulsar.common.functions.Resources;
 import org.apache.pulsar.common.functions.UpdateOptions;
 import org.apache.pulsar.common.io.SinkConfig;
-import org.apache.pulsar.functions.utils.FunctionCommon;
+import org.apache.pulsar.common.util.ClassLoaderUtils;
 import org.powermock.api.mockito.PowerMockito;
 import org.powermock.core.classloader.annotations.PowerMockIgnore;
 import org.powermock.core.classloader.annotations.PrepareForTest;
@@ -121,7 +121,7 @@ public void setup() throws Exception {
             throw new RuntimeException("Failed to file required test archive: " + JAR_FILE_NAME);
         }
         JAR_FILE_PATH = file.getFile();
-        Thread.currentThread().setContextClassLoader(FunctionCommon.loadJar(new File(JAR_FILE_PATH)));
+        Thread.currentThread().setContextClassLoader(ClassLoaderUtils.loadJar(new File(JAR_FILE_PATH)));
     }
 
     public SinkConfig getSinkConfig() {

pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdSources.java

@@ -40,7 +40,7 @@
 import org.apache.pulsar.common.functions.Resources;
 import org.apache.pulsar.common.functions.UpdateOptions;
 import org.apache.pulsar.common.io.SourceConfig;
-import org.apache.pulsar.functions.utils.FunctionCommon;
+import org.apache.pulsar.common.util.ClassLoaderUtils;
 import org.powermock.api.mockito.PowerMockito;
 import org.powermock.core.classloader.annotations.PowerMockIgnore;
 import org.powermock.core.classloader.annotations.PrepareForTest;
@@ -98,7 +98,7 @@ public void setup() throws Exception {
         mockStatic(CmdFunctions.class);
         PowerMockito.doNothing().when(localSourceRunner).runCmd();
         JAR_FILE_PATH = Thread.currentThread().getContextClassLoader().getResource(JAR_FILE_NAME).getFile();
-        Thread.currentThread().setContextClassLoader(FunctionCommon.loadJar(new File(JAR_FILE_PATH)));
+        Thread.currentThread().setContextClassLoader(ClassLoaderUtils.loadJar(new File(JAR_FILE_PATH)));
     }
 
     public SourceConfig getSourceConfig() {

pulsar-common/src/main/java/org/apache/pulsar/common/util/ClassLoaderUtils.java

@@ -0,0 +1,80 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.util;
+
+import java.io.File;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.nio.file.Path;
+
+/**
+ * Helper methods wrt Classloading.
+ */
+public class ClassLoaderUtils {
+    /**
+     * Load a jar.
+     *
+     * @param jar file of jar
+     * @return classloader
+     * @throws MalformedURLException
+     */
+    public static ClassLoader loadJar(File jar) throws MalformedURLException {
+        java.net.URL url = jar.toURI().toURL();
+        return new URLClassLoader(new URL[]{url});
+    }
+
+    public static ClassLoader extractClassLoader(Path archivePath, File packageFile) throws Exception {
+        if (archivePath != null) {
+            return loadJar(archivePath.toFile());
+        }
+        if (packageFile != null) {
+            return loadJar(packageFile);
+        }
+        return null;
+    }
+
+    public static Class<?> loadClass(String className, ClassLoader classLoader) throws ClassNotFoundException {
+        Class<?> objectClass;
+        try {
+            objectClass = Class.forName(className);
+        } catch (ClassNotFoundException | NoClassDefFoundError e) {
+            if (classLoader != null) {
+                objectClass = classLoader.loadClass(className);
+            } else {
+                throw e;
+            }
+        }
+        return objectClass;
+    }
+
+    public static void implementsClass(String className, Class<?> klass, ClassLoader classLoader) {
+        Class<?> objectClass;
+        try {
+            objectClass = loadClass(className, classLoader);
+        } catch (ClassNotFoundException | NoClassDefFoundError e) {
+            throw new IllegalArgumentException("Cannot find/load class " + className);
+        }
+
+        if (!klass.isAssignableFrom(objectClass)) {
+            throw new IllegalArgumentException(
+                    String.format("%s does not implement %s", className, klass.getName()));
+        }
+    }
+}

pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/Reflections.java → pulsar-common/src/main/java/org/apache/pulsar/common/util/Reflections.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.functions.utils;
+package org.apache.pulsar.common.util;
 
 import java.io.IOException;
 import java.lang.reflect.Array;
@@ -165,7 +165,7 @@ public static Object createInstance(String userClassName,
 
     public static Object createInstance(String userClassName, java.io.File jar) {
         try {
-            return createInstance(userClassName, FunctionCommon.loadJar(jar));
+            return createInstance(userClassName, ClassLoaderUtils.loadJar(jar));
         } catch (Exception ex) {
             return null;
         }
@@ -180,7 +180,7 @@ public static Object createInstance(String userClassName, java.io.File jar) {
      */
     public static boolean classExistsInJar(java.io.File jar, String fqcn) {
         try {
-            java.net.URLClassLoader loader = (URLClassLoader) FunctionCommon.loadJar(jar);
+            java.net.URLClassLoader loader = (URLClassLoader) ClassLoaderUtils.loadJar(jar);
             Class.forName(fqcn, false, loader);
             loader.close();
             return true;
@@ -199,7 +199,7 @@ public static boolean classExists(String fqcn) {
         try {
             Class.forName(fqcn);
             return true;
-        } catch( ClassNotFoundException e ) {
+        } catch (ClassNotFoundException e) {
            return false;
         }
     }
@@ -214,7 +214,7 @@ public static boolean classExists(String fqcn) {
     public static boolean classInJarImplementsIface(java.io.File jar, String fqcn, Class xface) {
         boolean ret = false;
         try {
-            java.net.URLClassLoader loader = (URLClassLoader) FunctionCommon.loadJar(jar);
+            java.net.URLClassLoader loader = (URLClassLoader) ClassLoaderUtils.loadJar(jar);
             if (xface.isAssignableFrom(Class.forName(fqcn, false, loader))){
                 ret = true;
             }
@@ -281,7 +281,7 @@ public static Class loadClass(String className, ClassLoader classLoader) throws
                 throw new ClassNotFoundException(className);
             }
         } else if (isPrimitive(className)) {
-            return (Class)PRIMITIVE_NAME_TYPE_MAP.get(className);
+            return (Class) PRIMITIVE_NAME_TYPE_MAP.get(className);
         } else if (className.charAt(0) == 'L' && className.charAt(className.length() - 1) == ';') {
             return classLoader.loadClass(className.substring(1, className.length() - 1));
         } else {
@@ -291,10 +291,12 @@ public static Class loadClass(String className, ClassLoader classLoader) throws
                 if (className.charAt(0) != '[') {
                     throw var4;
                 } else {
+                    // CHECKSTYLE.OFF: EmptyStatement
                     int arrayDimension;
-                    for(arrayDimension = 0; className.charAt(arrayDimension) == '['; ++arrayDimension) {
+                    for (arrayDimension = 0; className.charAt(arrayDimension) == '['; ++arrayDimension) {
                         ;
                     }
+                    // CHECKSTYLE.ON: EmptyStatement
 
                     Class componentType = loadClass(className.substring(arrayDimension), classLoader);
                     return Array.newInstance(componentType, new int[arrayDimension]).getClass();

pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/ReflectionsTest.java → pulsar-common/src/test/java/org/apache/pulsar/common/util/ReflectionsTest.java

@@ -16,15 +16,16 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.functions.utils;
+package org.apache.pulsar.common.util;
 
-import static org.apache.pulsar.functions.utils.Reflections.createInstance;
+import static org.apache.pulsar.common.util.Reflections.createInstance;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertTrue;
 import static org.testng.AssertJUnit.fail;
 
 import java.lang.reflect.InvocationTargetException;
+
 import org.testng.annotations.Test;
 
 /**

pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/InstanceUtils.java

@@ -28,7 +28,7 @@
 import org.apache.pulsar.functions.api.SerDe;
 import org.apache.pulsar.functions.proto.Function;
 import org.apache.pulsar.functions.sink.PulsarSink;
-import org.apache.pulsar.functions.utils.Reflections;
+import org.apache.pulsar.common.util.Reflections;
 
 import net.jodah.typetools.TypeResolver;
 import org.apache.pulsar.functions.utils.FunctionCommon;

pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java

@@ -27,7 +27,7 @@
 
 import java.io.IOException;
 import java.util.concurrent.CompletableFuture;
-import lombok.AccessLevel;
+
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import net.jodah.typetools.TypeResolver;
@@ -69,7 +69,7 @@
 import org.apache.pulsar.functions.sink.PulsarSinkDisable;
 import org.apache.pulsar.functions.source.PulsarSource;
 import org.apache.pulsar.functions.source.PulsarSourceConfig;
-import org.apache.pulsar.functions.utils.Reflections;
+import org.apache.pulsar.common.util.Reflections;
 import org.apache.pulsar.functions.utils.FunctionCommon;
 import org.apache.pulsar.functions.utils.functioncache.FunctionCacheManager;
 import org.apache.pulsar.io.core.Sink;

pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java

@@ -40,7 +40,7 @@
 import org.apache.pulsar.functions.instance.stats.ComponentStatsManager;
 import org.apache.pulsar.functions.source.PulsarRecord;
 import org.apache.pulsar.functions.source.TopicSchema;
-import org.apache.pulsar.functions.utils.Reflections;
+import org.apache.pulsar.common.util.Reflections;
 import org.apache.pulsar.io.core.Sink;
 import org.apache.pulsar.io.core.SinkContext;
 

pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSource.java

@@ -23,7 +23,6 @@
 import com.google.common.annotations.VisibleForTesting;
 
 import java.util.*;
-import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
@@ -35,7 +34,7 @@
 import org.apache.pulsar.client.impl.MultiTopicsConsumerImpl;
 import org.apache.pulsar.functions.api.Record;
 import org.apache.pulsar.common.functions.FunctionConfig;
-import org.apache.pulsar.functions.utils.Reflections;
+import org.apache.pulsar.common.util.Reflections;
 import org.apache.pulsar.io.core.PushSource;
 import org.apache.pulsar.io.core.SourceContext;
 

pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/windowing/WindowFunctionExecutor.java

@@ -31,7 +31,7 @@
 import lombok.extern.slf4j.Slf4j;
 
 import org.apache.pulsar.functions.api.*;
-import org.apache.pulsar.functions.utils.Reflections;
+import org.apache.pulsar.common.util.Reflections;
 import org.apache.pulsar.common.functions.WindowConfig;
 import org.apache.pulsar.functions.windowing.evictors.CountEvictionPolicy;
 import org.apache.pulsar.functions.windowing.evictors.TimeEvictionPolicy;

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/auth/FunctionAuthProvider.java

@@ -21,7 +21,7 @@
 import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
 import org.apache.pulsar.functions.instance.AuthenticationConfig;
 import org.apache.pulsar.functions.proto.Function;
-import org.apache.pulsar.functions.utils.Reflections;
+import org.apache.pulsar.common.util.Reflections;
 
 import java.util.Optional;
 

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/auth/KubernetesFunctionAuthProvider.java

@@ -19,11 +19,9 @@
 package org.apache.pulsar.functions.auth;
 
 import io.kubernetes.client.apis.CoreV1Api;
-import io.kubernetes.client.models.V1ServiceAccount;
 import io.kubernetes.client.models.V1StatefulSet;
-import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
 import org.apache.pulsar.functions.proto.Function;
-import org.apache.pulsar.functions.utils.Reflections;
+import org.apache.pulsar.common.util.Reflections;
 
 import java.util.Optional;
 

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/JavaInstanceStarter.java

@@ -51,7 +51,7 @@
 import org.apache.pulsar.functions.runtime.thread.ThreadRuntimeFactory;
 import org.apache.pulsar.functions.secretsprovider.ClearTextSecretsProvider;
 import org.apache.pulsar.functions.secretsprovider.SecretsProvider;
-import org.apache.pulsar.functions.utils.Reflections;
+import org.apache.pulsar.common.util.Reflections;
 
 import javax.management.MalformedObjectNameException;
 import java.lang.reflect.Type;

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeCustomizer.java

@@ -18,7 +18,7 @@
  */
 package org.apache.pulsar.functions.runtime;
 
-import org.apache.pulsar.functions.utils.Reflections;
+import org.apache.pulsar.common.util.Reflections;
 
 import java.util.Map;
 

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeFactory.java

@@ -24,7 +24,7 @@
 import org.apache.pulsar.functions.instance.InstanceConfig;
 import org.apache.pulsar.functions.proto.Function;
 import org.apache.pulsar.functions.secretsproviderconfigurator.SecretsProviderConfigurator;
-import org.apache.pulsar.functions.utils.Reflections;
+import org.apache.pulsar.common.util.Reflections;
 import org.apache.pulsar.functions.worker.WorkerConfig;
 
 import java.util.Optional;

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/thread/ThreadRuntimeFactory.java

@@ -36,7 +36,7 @@
 import org.apache.pulsar.functions.runtime.RuntimeUtils;
 import org.apache.pulsar.functions.secretsprovider.SecretsProvider;
 import org.apache.pulsar.functions.secretsproviderconfigurator.SecretsProviderConfigurator;
-import org.apache.pulsar.functions.utils.Reflections;
+import org.apache.pulsar.common.util.Reflections;
 import org.apache.pulsar.functions.utils.functioncache.FunctionCacheManager;
 import org.apache.pulsar.functions.utils.functioncache.FunctionCacheManagerImpl;
 import org.apache.pulsar.functions.worker.WorkerConfig;