This is an automated email from the ASF dual-hosted git repository. eolivelli pushed a commit to branch branch-2.9 in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 64a63bb2d198757b95b13bfc72090a30902a231b Author: Enrico Olivelli <[email protected]> AuthorDate: Tue Nov 9 12:07:29 2021 +0100 Pulsar Functions: detect .nar files and prevent spammy logs on functions boot (#13) (#12667) (cherry picked from commit 515a69fc6675322c455d606631cc47490756b453) --- .../org/apache/pulsar/common/nar/FileUtils.java | 20 ++++++++++++++++ .../functions/instance/JavaInstanceRunnable.java | 1 + .../functions/runtime/thread/ThreadRuntime.java | 28 +++++++++++++++------- 3 files changed, 41 insertions(+), 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 index cc67730..0bfdb80 100644 --- 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,6 +30,9 @@ import java.io.FilenameFilter; 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; /** @@ -37,6 +40,7 @@ import org.slf4j.Logger; * operations. * */ +@Slf4j public class FileUtils { public static final long MILLIS_BETWEEN_ATTEMPTS = 50L; @@ -221,5 +225,21 @@ public class FileUtils { /* 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/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java index cfdcb08..bbc7ced 100644 --- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java +++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java @@ -53,6 +53,7 @@ import org.apache.pulsar.common.functions.FunctionConfig; import org.apache.pulsar.common.functions.ProducerConfig; import org.apache.pulsar.common.util.ObjectMapperFactory; 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; diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/thread/ThreadRuntime.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/thread/ThreadRuntime.java index b6dd019..be43bb0 100644 --- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/thread/ThreadRuntime.java +++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/thread/ThreadRuntime.java @@ -19,6 +19,7 @@ package org.apache.pulsar.functions.runtime.thread; +import java.io.File; import java.io.FileNotFoundException; import java.io.IOException; import java.util.Arrays; @@ -33,6 +34,7 @@ import org.apache.pulsar.broker.PulsarServerException; import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.api.ClientBuilder; import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.common.nar.FileUtils; import org.apache.pulsar.functions.instance.InstanceConfig; import org.apache.pulsar.functions.instance.InstanceUtils; import org.apache.pulsar.functions.instance.stats.FunctionCollectorRegistry; @@ -132,14 +134,24 @@ public class ThreadRuntime implements Runtime { return Thread.currentThread().getContextClassLoader(); } 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(),
