Added catch for NoClassDefFoundError wherever there was a ClassNotFoundException (#5870)

Fixes #5726 

### Motivation

When running pulsar-io connectors and functions from the Intellij IDE some actions fail
due to uncaught class-not-found throwables.
The expectation being that the class is being dynamically loaded and only the ClassNotFoundException will occur if the class is not found.
When the function is created or run with https://pulsar.apache.org/docs/en/functions-deploying/#local-run-mode this is indeed the case.
When running under the control https://pulsar.apache.org/docs/en/functions-debug/#debug-with-localrun-mode as a gradle plugin the class may already be known and throw a NoClassDefFoundError.
It seems to me that any time ClassNotFoundException is handled then NoClassDefFoundError should also be caught.

### Modifications

Wherever there was a `catch (ClassNotFoundException ` I replaced it with 
`catch (ClassNotFoundException | NoClassDefFoundError ` .
There were multiple cases where the ClassNotFoundException were handled e.g. the jar loader failed so the nar loader was used to handle the jar loader's failure.

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

@@ -1416,7 +1416,7 @@ public class BrokerService implements Closeable, ZooKeeperCacheListener<Policies
         addDynamicConfigValidator("loadManagerClassName", (className) -> {
             try {
                 Class.forName(className);
-            } catch (ClassNotFoundException e) {
+            } catch (ClassNotFoundException | NoClassDefFoundError e) {
                 log.warn("Configured load-manager class {} not found {}", className, e.getMessage());
                 return false;
             }

pulsar-client-api/src/main/java/org/apache/pulsar/client/internal/ReflectionUtils.java

@@ -59,7 +59,7 @@ class ReflectionUtils {
                 // that loaded the API
                 return (Class<T>) Thread.currentThread().getContextClassLoader().loadClass(className);
             }
-        } catch (ClassNotFoundException e) {
+        } catch (ClassNotFoundException | NoClassDefFoundError e) {
             throw new RuntimeException(e);
         }
     }

pulsar-client-kafka-compat/pulsar-client-kafka_0_8/src/main/java/org/apache/kafka/clients/producer/PulsarKafkaProducer.java

@@ -228,7 +228,7 @@ public class PulsarKafkaProducer<K, V> extends Producer<K, V> {
         Class<?> c = null;
         try {
             c = Class.forName(key);
-        } catch (ClassNotFoundException e) {
+        } catch (ClassNotFoundException | NoClassDefFoundError e) {
             throw new IllegalArgumentException("class not found for :" + key);
         }
         if (c == null)

pulsar-common/src/main/java/org/apache/pulsar/common/nar/NarClassLoader.java

@@ -141,7 +141,7 @@ public class NarClassLoader extends URLClassLoader {
         File unpacked = NarUnpacker.unpackNar(narPath, NAR_CACHE_DIR);
         try {
             return new NarClassLoader(unpacked, additionalJars, NarClassLoader.class.getClassLoader());
-        } catch (ClassNotFoundException e) {
+        } catch (ClassNotFoundException | NoClassDefFoundError e) {
             throw new IOException(e);
         }
     }
@@ -151,7 +151,7 @@ public class NarClassLoader extends URLClassLoader {
         File unpacked = NarUnpacker.unpackNar(narPath, NAR_CACHE_DIR);
         try {
             return new NarClassLoader(unpacked, additionalJars, parent);
-        } catch (ClassNotFoundException e) {
+        } catch (ClassNotFoundException | NoClassDefFoundError e) {
             throw new IOException(e);
         }
     }

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

@@ -162,7 +162,7 @@ public class TopicSchema {
         try {
             Class<?> protobufBaseClass = Class.forName("com.google.protobuf.GeneratedMessageV3");
             return protobufBaseClass.isAssignableFrom(pojoClazz);
-        } catch (ClassNotFoundException e) {
+        } catch (ClassNotFoundException | NoClassDefFoundError e) {
             // If function does not have protobuf in classpath then it cannot be protobuf
             return false;
         }

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

@@ -131,7 +131,7 @@ public class WindowFunctionExecutor<I, O> implements Function<I, O> {
         try {
             theCls = Class.forName(windowConfig.getTimestampExtractorClassName(),
                     true, Thread.currentThread().getContextClassLoader());
-        } catch (ClassNotFoundException cnfe) {
+        } catch (ClassNotFoundException | NoClassDefFoundError cnfe) {
             throw new RuntimeException(
                     String.format("Timestamp extractor class %s must be in class path",
                             windowConfig.getTimestampExtractorClassName()), cnfe);

pulsar-functions/runtime-all/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceMain.java

@@ -104,7 +104,7 @@ public class JavaInstanceMain {
         Class<?> theCls;
         try {
             theCls = Class.forName(userClassName, true, classLoader);
-        } catch (ClassNotFoundException cnfe) {
+        } catch (ClassNotFoundException | NoClassDefFoundError cnfe) {
             throw new RuntimeException("Class " + userClassName + " must be in class path", cnfe);
         }
         Object result;

pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionCommon.java

@@ -123,10 +123,10 @@ public class FunctionCommon {
         Class<?> theCls;
         try {
             theCls = Class.forName(userClassName);
-        } catch (ClassNotFoundException cnfe) {
+        } catch (ClassNotFoundException | NoClassDefFoundError cnfe) {
             try {
                 theCls = Class.forName(userClassName, true, classLoader);
-            } catch (ClassNotFoundException e) {
+            } catch (ClassNotFoundException | NoClassDefFoundError e) {
                 throw new RuntimeException("User class must be in class path", cnfe);
             }
         }
@@ -274,7 +274,7 @@ public class FunctionCommon {
         Class<?> objectClass;
         try {
             objectClass = loadClass(className, classLoader);
-        } catch (ClassNotFoundException e) {
+        } catch (ClassNotFoundException | NoClassDefFoundError e) {
             throw new IllegalArgumentException("Cannot find/load class " + className);
         }
 
@@ -288,7 +288,7 @@ public class FunctionCommon {
         Class<?> objectClass;
         try {
             objectClass = Class.forName(className);
-        } catch (ClassNotFoundException e) {
+        } catch (ClassNotFoundException | NoClassDefFoundError e) {
             if (classLoader != null) {
                 objectClass = classLoader.loadClass(className);
             } else {

pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java

@@ -52,7 +52,7 @@ public class FunctionConfigUtils {
             if (classLoader != null) {
                 try {
                     typeArgs = FunctionCommon.getFunctionTypes(functionConfig, classLoader);
-                } catch (ClassNotFoundException e) {
+                } catch (ClassNotFoundException | NoClassDefFoundError e) {
                     throw new IllegalArgumentException(
                             String.format("Function class %s must be in class path", functionConfig.getClassName()), e);
                 }
@@ -386,7 +386,7 @@ public class FunctionConfigUtils {
                         String.format("Function class %s does not implement the correct interface",
                                 functionClass.getName()));
             }
-        } catch (ClassNotFoundException e) {
+        } catch (ClassNotFoundException | NoClassDefFoundError e) {
             throw new IllegalArgumentException(
                     String.format("Function class %s must be in class path", functionConfig.getClassName()), e);
         }
@@ -394,7 +394,7 @@ public class FunctionConfigUtils {
         Class<?>[] typeArgs;
         try {
             typeArgs = FunctionCommon.getFunctionTypes(functionConfig, clsLoader);
-        } catch (ClassNotFoundException e) {
+        } catch (ClassNotFoundException | NoClassDefFoundError e) {
             throw new IllegalArgumentException(
                     String.format("Function class %s must be in class path", functionConfig.getClassName()), e);
         }

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

@@ -63,7 +63,7 @@ public class Reflections {
         Class<?> theCls;
         try {
             theCls = Class.forName(userClassName, true, classLoader);
-        } catch (ClassNotFoundException cnfe) {
+        } catch (ClassNotFoundException | NoClassDefFoundError cnfe) {
             throw new RuntimeException("User class must be in class path", cnfe);
         }
         if (!xface.isAssignableFrom(theCls)) {
@@ -104,7 +104,7 @@ public class Reflections {
         Class<?> theCls;
         try {
             theCls = Class.forName(userClassName, true, classLoader);
-        } catch (ClassNotFoundException cnfe) {
+        } catch (ClassNotFoundException | NoClassDefFoundError cnfe) {
             throw new RuntimeException("User class must be in class path", cnfe);
         }
         Object result;
@@ -138,7 +138,7 @@ public class Reflections {
         Class<?> theCls;
         try {
             theCls = Class.forName(userClassName, true, classLoader);
-        } catch (ClassNotFoundException cnfe) {
+        } catch (ClassNotFoundException | NoClassDefFoundError cnfe) {
             throw new RuntimeException("User class must be in class path", cnfe);
         }
         Object result;
@@ -184,7 +184,7 @@ public class Reflections {
             Class.forName(fqcn, false, loader);
             loader.close();
             return true;
-        } catch (ClassNotFoundException | IOException e) {
+        } catch (ClassNotFoundException | NoClassDefFoundError | IOException e) {
             return false;
         }
     }
@@ -219,7 +219,7 @@ public class Reflections {
                 ret = true;
             }
             loader.close();
-        } catch (ClassNotFoundException | IOException e) {
+        } catch (ClassNotFoundException | NoClassDefFoundError | IOException e) {
             throw new RuntimeException(e);
         }
         return ret;
@@ -238,7 +238,7 @@ public class Reflections {
             if (xface.isAssignableFrom(Class.forName(fqcn))){
                 ret = true;
             }
-        } catch (ClassNotFoundException e) {
+        } catch (ClassNotFoundException | NoClassDefFoundError e) {
             throw new RuntimeException(e);
         }
         return ret;
@@ -287,7 +287,7 @@ public class Reflections {
         } else {
             try {
                 return classLoader.loadClass(className);
-            } catch (ClassNotFoundException var4) {
+            } catch (ClassNotFoundException | NoClassDefFoundError var4) {
                 if (className.charAt(0) != '[') {
                     throw var4;
                 } else {

pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java

@@ -377,7 +377,7 @@ public class SinkConfigUtils {
             try {
                 typeArg = getSinkType(sinkClassName, narClassLoader);
                 classLoader = narClassLoader;
-            } catch (ClassNotFoundException e) {
+            } catch (ClassNotFoundException | NoClassDefFoundError e) {
                 throw new IllegalArgumentException(
                         String.format("Sink class %s must be in class path", sinkClassName), e);
             }
@@ -388,13 +388,13 @@ public class SinkConfigUtils {
                 try {
                     typeArg = getSinkType(sinkClassName, jarClassLoader);
                     classLoader = jarClassLoader;
-                } catch (ClassNotFoundException e) {
+                } catch (ClassNotFoundException | NoClassDefFoundError e) {
                     // class not found in JAR try loading as a NAR and searching for the class
                     if (narClassLoader != null) {
                         try {
                             typeArg = getSinkType(sinkClassName, narClassLoader);
                             classLoader = narClassLoader;
-                        } catch (ClassNotFoundException e1) {
+                        } catch (ClassNotFoundException | NoClassDefFoundError e1) {
                             throw new IllegalArgumentException(
                                     String.format("Sink class %s must be in class path", sinkClassName), e1);
                         }
@@ -407,7 +407,7 @@ public class SinkConfigUtils {
                 try {
                     typeArg = getSinkType(sinkClassName, narClassLoader);
                     classLoader = narClassLoader;
-                } catch (ClassNotFoundException e1) {
+                } catch (ClassNotFoundException | NoClassDefFoundError e1) {
                     throw new IllegalArgumentException(
                             String.format("Sink class %s must be in class path", sinkClassName), e1);
                 }

pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java

@@ -271,7 +271,7 @@ public class SourceConfigUtils {
             try {
                 typeArg = getSourceType(sourceClassName, narClassLoader);
                 classLoader = narClassLoader;
-            } catch (ClassNotFoundException e) {
+            } catch (ClassNotFoundException | NoClassDefFoundError e) {
                 throw new IllegalArgumentException(
                         String.format("Source class %s must be in class path", sourceClassName), e);
             }
@@ -282,13 +282,13 @@ public class SourceConfigUtils {
                 try {
                     typeArg = getSourceType(sourceClassName, jarClassLoader);
                     classLoader = jarClassLoader;
-                } catch (ClassNotFoundException e) {
+                } catch (ClassNotFoundException | NoClassDefFoundError e) {
                     // class not found in JAR try loading as a NAR and searching for the class
                     if (narClassLoader != null) {
                         try {
                             typeArg = getSourceType(sourceClassName, narClassLoader);
                             classLoader = narClassLoader;
-                        } catch (ClassNotFoundException e1) {
+                        } catch (ClassNotFoundException | NoClassDefFoundError e1) {
                             throw new IllegalArgumentException(
                                     String.format("Source class %s must be in class path", sourceClassName), e1);
                         }
@@ -301,7 +301,7 @@ public class SourceConfigUtils {
                 try {
                     typeArg = getSourceType(sourceClassName, narClassLoader);
                     classLoader = narClassLoader;
-                } catch (ClassNotFoundException e1) {
+                } catch (ClassNotFoundException | NoClassDefFoundError e1) {
                     throw new IllegalArgumentException(
                             String.format("Source class %s must be in class path", sourceClassName), e1);
                 }

pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/ValidatorUtils.java

@@ -63,7 +63,7 @@ public class ValidatorUtils {
         if (inputSerializer.equals(DEFAULT_SERDE)) return;
         try {
             Class<?> serdeClass = FunctionCommon.loadClass(inputSerializer, clsLoader);
-        } catch (ClassNotFoundException e) {
+        } catch (ClassNotFoundException | NoClassDefFoundError e) {
             throw new IllegalArgumentException(
                     String.format("The input serialization/deserialization class %s does not exist",
                             inputSerializer));
@@ -84,7 +84,7 @@ public class ValidatorUtils {
         try {
             fnInputClass = Class.forName(typeArg.getName(), true, clsLoader);
             serdeInputClass = Class.forName(serDeTypes[0].getName(), true, clsLoader);
-        } catch (ClassNotFoundException e) {
+        } catch (ClassNotFoundException | NoClassDefFoundError e) {
             throw new IllegalArgumentException("Failed to load type class", e);
         }
 
@@ -115,7 +115,7 @@ public class ValidatorUtils {
         try {
             fnInputClass = Class.forName(typeArg.getName(), true, clsLoader);
             schemaInputClass = Class.forName(schemaTypes[0].getName(), true, clsLoader);
-        } catch (ClassNotFoundException e) {
+        } catch (ClassNotFoundException | NoClassDefFoundError e) {
             throw new IllegalArgumentException("Failed to load type class", e);
         }
 
@@ -147,7 +147,7 @@ public class ValidatorUtils {
         Class functionClass;
         try {
             functionClass = classLoader.loadClass(functionDetailsBuilder.getClassName());
-        } catch (ClassNotFoundException e) {
+        } catch (ClassNotFoundException | NoClassDefFoundError e) {
             throw new IllegalArgumentException(
                     String.format("Function class %s must be in class path", functionDetailsBuilder.getClassName()), e);
         }

pulsar-io/flume/src/main/java/org/apache/pulsar/io/flume/node/PropertiesFileConfigurationProvider.java

@@ -196,7 +196,7 @@ public class PropertiesFileConfigurationProvider extends
         } catch (IOException ex) {
             LOGGER.error("Unable to load file:" + file
                     + " (I/O failure) - Exception follows.", ex);
-        } catch (ClassNotFoundException e) {
+        } catch (ClassNotFoundException | NoClassDefFoundError e) {
             LOGGER.error("Configuration resolver class not found", e);
         } catch (InstantiationException e) {
             LOGGER.error("Instantiation exception", e);

pulsar-io/hdfs2/src/main/java/org/apache/pulsar/io/hdfs2/AbstractHdfsConnector.java

@@ -237,7 +237,7 @@ public abstract class AbstractHdfsConnector {
             if (clazz == null) {
                 try {
                     clazz = Class.forName(name, true, classLoader);
-                } catch (ClassNotFoundException e) {
+                } catch (ClassNotFoundException | NoClassDefFoundError e) {
                     return null;
                 }
                 // two putters can race here, but they'll put the same class

pulsar-io/hdfs3/src/main/java/org/apache/pulsar/io/hdfs3/AbstractHdfsConnector.java

@@ -237,7 +237,7 @@ public abstract class AbstractHdfsConnector {
             if (clazz == null) {
                 try {
                     clazz = Class.forName(name, true, classLoader);
-                } catch (ClassNotFoundException e) {
+                } catch (ClassNotFoundException | NoClassDefFoundError e) {
                     return null;
                 }
                 // two putters can race here, but they'll put the same class

pulsar-io/jdbc/src/test/java/org/apache/pulsar/io/jdbc/SqliteUtils.java

@@ -37,7 +37,7 @@ public final class SqliteUtils {
     static {
         try {
             Class.forName("org.sqlite.JDBC");
-        } catch (ClassNotFoundException e) {
+        } catch (ClassNotFoundException | NoClassDefFoundError e) {
             throw new RuntimeException(e);
         }
     }

pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorUtils.java

@@ -56,7 +56,7 @@ public class PulsarConnectorUtils {
         Class<?> theCls;
         try {
             theCls = Class.forName(userClassName, true, classLoader);
-        } catch (ClassNotFoundException cnfe) {
+        } catch (ClassNotFoundException | NoClassDefFoundError cnfe) {
             throw new RuntimeException("User class must be in class path", cnfe);
         }
         if (!xface.isAssignableFrom(theCls)) {