Pulsar Functions: detect .nar files and prevent spammy logs on functi…

…ons boot (#12665)
apache · Nov 9, 2021 · 48ee424 · 48ee424
1 parent 81fe6b6
commit 48ee424
Show file tree

Hide file tree

Showing 2 changed files with 41 additions and 8 deletions.
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/nar/FileUtils.java b/pulsar-common/src/main/java/org/apache/pulsar/common/nar/FileUtils.java
@@ -30,14 +30,18 @@
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipFile;
 
+import lombok.extern.slf4j.Slf4j;
 import org.slf4j.Logger;
 
 /**
  * A utility class containing a few useful static methods to do typical IO
  * operations.
  *
  */
+@Slf4j
 public class FileUtils {
 
     public static final long MILLIS_BETWEEN_ATTEMPTS = 50L;
@@ -221,5 +225,21 @@ public static void sleepQuietly(final long millis) {
             /* do nothing */
         }
     }
+
+    public static boolean mayBeANarArchive(File jarFile) {
+        try (ZipFile zipFile = new ZipFile(jarFile);) {
+            ZipEntry entry = zipFile.getEntry("META-INF/bundled-dependencies");
+            if (entry == null || !entry.isDirectory()) {
+                log.info("Jar file {} does not contain META-INF/bundled-dependencies, it is not a NAR file", jarFile);
+                return false;
+            } else {
+                log.info("Jar file {} contains META-INF/bundled-dependencies, it may be a NAR file", jarFile);
+                return true;
+            }
+        } catch (IOException err) {
+            log.info("Cannot safely detect if {} is a NAR archive", jarFile, err);
+            return true;
+        }
+    }
 }
 
diff --git a/...ons/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java b/...ons/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
@@ -22,6 +22,8 @@
 import com.google.gson.Gson;
 import com.google.gson.reflect.TypeToken;
 import io.prometheus.client.CollectorRegistry;
+
+import java.io.File;
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.util.Arrays;
@@ -44,6 +46,7 @@
 import org.apache.pulsar.common.functions.FunctionConfig;
 import org.apache.pulsar.common.functions.ProducerConfig;
 import org.apache.pulsar.common.util.Reflections;
+import org.apache.pulsar.common.nar.FileUtils;
 import org.apache.pulsar.functions.api.Function;
 import org.apache.pulsar.functions.api.Record;
 import org.apache.pulsar.functions.api.StateStore;
@@ -292,14 +295,24 @@ public void run() {
 
     private ClassLoader loadJars() throws Exception {
         ClassLoader fnClassLoader;
-        try {
-            log.info("Load JAR: {}", jarFile);
-            // Let's first try to treat it as a nar archive
-            fnCache.registerFunctionInstanceWithArchive(
-                instanceConfig.getFunctionId(),
-                instanceConfig.getInstanceName(),
-                jarFile, narExtractionDirectory);
-        } catch (FileNotFoundException e) {
+        boolean loadedAsNar = false;
+        if (FileUtils.mayBeANarArchive(new File(jarFile))) {
+            try {
+                log.info("Trying Loading file as NAR file: {}", jarFile);
+                // Let's first try to treat it as a nar archive
+                fnCache.registerFunctionInstanceWithArchive(
+                        instanceConfig.getFunctionId(),
+                        instanceConfig.getInstanceName(),
+                        jarFile, narExtractionDirectory);
+                loadedAsNar = true;
+            } catch (FileNotFoundException e) {
+                // this is usually like
+                // java.io.FileNotFoundException: /tmp/pulsar-nar/xxx.jar-unpacked/xxxxx/META-INF/MANIFEST.MF'
+                log.error("The file {} does not look like a .nar file", jarFile, e.toString());
+            }
+        }
+        if (!loadedAsNar) {
+            log.info("Load file as simple JAR file: {}", jarFile);
             // create the function class loader
             fnCache.registerFunctionInstance(
                     instanceConfig.getFunctionId(),