[broker] EntryFilters fix NoClassDefFoundError due to closed classloader

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/plugin/EntryFilterProvider.java

@@ -197,7 +197,8 @@ protected EntryFilter load(EntryFilterMetaData metadata)
                         + " does not implement entry filter interface");
             }
             EntryFilter pi = (EntryFilter) filter;
-            return new EntryFilterWithClassLoader(pi, ncl);
+            // the classloader is shared with the broker, the instance doesn't own it
+            return new EntryFilterWithClassLoader(pi, ncl, false);
         } catch (Throwable e) {
             if (e instanceof IOException) {
                 throw (IOException) e;

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/plugin/EntryFilterWithClassLoader.java

@@ -23,22 +23,27 @@
 import lombok.ToString;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.bookkeeper.mledger.Entry;
+import org.apache.pulsar.broker.ClassLoaderSwitcher;
 import org.apache.pulsar.common.nar.NarClassLoader;
 
 @Slf4j
 @ToString
 public class EntryFilterWithClassLoader implements EntryFilter {
     private final EntryFilter entryFilter;
     private final NarClassLoader classLoader;
+    private final boolean classLoaderOwned;
 
-    public EntryFilterWithClassLoader(EntryFilter entryFilter, NarClassLoader classLoader) {
+    public EntryFilterWithClassLoader(EntryFilter entryFilter, NarClassLoader classLoader, boolean classLoaderOwned) {
         this.entryFilter = entryFilter;
         this.classLoader = classLoader;
+        this.classLoaderOwned = classLoaderOwned;
     }
 
     @Override
     public FilterResult filterEntry(Entry entry, FilterContext context) {
-        return entryFilter.filterEntry(entry, context);
+        try (ClassLoaderSwitcher switcher = new ClassLoaderSwitcher(classLoader)) {
+            return entryFilter.filterEntry(entry, context);
+        }
     }
 
     @VisibleForTesting
@@ -48,11 +53,16 @@ public EntryFilter getEntryFilter() {
 
     @Override
     public void close() {
-        entryFilter.close();
-        try {
-            classLoader.close();
-        } catch (IOException e) {
-            log.error("close EntryFilterWithClassLoader failed", e);
+        try (ClassLoaderSwitcher switcher = new ClassLoaderSwitcher(classLoader)) {
+            entryFilter.close();
+        }
+        if (classLoaderOwned) {
+            log.info("Closing classloader {} for EntryFilter {}", classLoader, entryFilter.getClass().getName());
+            try {
+                classLoader.close();
+            } catch (IOException e) {
+                log.error("close EntryFilterWithClassLoader failed", e);
+            }
         }
     }
 }

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

@@ -40,6 +40,7 @@
 import java.util.Comparator;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 
@@ -135,6 +136,7 @@ public class NarClassLoader extends URLClassLoader {
      * The NAR for which this <tt>ClassLoader</tt> is responsible.
      */
     private final File narWorkingDirectory;
+    private final AtomicBoolean closed = new AtomicBoolean();
 
     private static final String TMP_DIR_PREFIX = "pulsar-nar";
 
@@ -292,4 +294,18 @@ protected String findLibrary(final String libname) {
     public String toString() {
         return NarClassLoader.class.getName() + "[" + narWorkingDirectory.getPath() + "]";
     }
+
+    @Override
+    protected Class<?> loadClass(String name, boolean resolve) throws ClassNotFoundException {
+        if (closed.get()) {
+            log.warn("Loading class {} from a closed classloader ({})", name, this);
+        }
+        return super.loadClass(name, resolve);
+    }
+
+    @Override
+    public void close() throws IOException {
+        closed.set(true);
+        super.close();
+    }
 }