feat: Separate legacy drivers into their own classpath, ref: NOISSUED

Author: mihkels

benchmark-framework/pom.xml

@@ -74,11 +74,6 @@
             <artifactId>driver-nsq</artifactId>
             <version>${project.version}</version>
         </dependency>
-        <dependency>
-            <groupId>${project.groupId}</groupId>
-            <artifactId>driver-pravega</artifactId>
-            <version>${project.version}</version>
-        </dependency>
         <dependency>
             <groupId>${project.groupId}</groupId>
             <artifactId>driver-pulsar</artifactId>
@@ -173,6 +168,12 @@
             <version>${project.version}</version>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>driver-pravega</artifactId>
+            <version>${project.version}</version>
+            <scope>provided</scope>
+        </dependency>
         <dependency>
             <groupId>com.github.stefanbirkner</groupId>
             <artifactId>system-lambda</artifactId>
@@ -201,7 +202,7 @@
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-dependency-plugin</artifactId>
-                <version>2.10</version>
+                <version>3.9.0</version>
                 <executions>
                     <execution>
                         <id>build-classpath</id>

benchmark-framework/src/main/java/io/openmessaging/benchmark/Benchmark.java

@@ -72,6 +72,11 @@ static class Arguments {
                 description = "Allocate extra consumer workers when your backlog builds.")
         boolean extraConsumers;
 
+        @Parameter(
+                names = {"--legacy-driver-name"},
+                description = "Path to legacy driver with conflicting or not maintained dependencies")
+        public String legacyDriverName;
+
         @Parameter(description = "Workloads") // , required = true)
         public List<String> workloads;
 
@@ -165,7 +170,16 @@ static void main(String[] args) throws Exception {
                                         // Stop any left over workload
                                         worker.stopAll();
 
-                                        worker.initializeDriver(new File(driverConfig));
+                                        File legacyDriverFile =
+                                                arguments.legacyDriverName != null
+                                                        ? new File(arguments.legacyDriverName)
+                                                        : null;
+
+                                        if (worker instanceof LocalWorker) {
+                                            worker.initializeDriver(new File(driverConfig), legacyDriverFile);
+                                        } else {
+                                            worker.initializeDriver(new File(driverConfig));
+                                        }
 
                                         WorkloadGenerator generator =
                                                 new WorkloadGenerator(driverConfiguration.name, workload, worker);

benchmark-framework/src/main/java/io/openmessaging/benchmark/utils/IsolatedDriverLoader.java

@@ -0,0 +1,127 @@
+/*
+ * Licensed 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 io.openmessaging.benchmark.utils;
+
+import java.io.File;
+import java.lang.reflect.Constructor;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.util.ArrayList;
+import java.util.List;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Child-first classloader that isolates driver dependencies from the main classpath. Use this to
+ * load drivers with conflicting transitive dependencies.
+ */
+public final class IsolatedDriverLoader extends URLClassLoader {
+    private static final Logger log = LoggerFactory.getLogger(IsolatedDriverLoader.class);
+
+    private IsolatedDriverLoader(URL[] urls, ClassLoader parent) {
+        super(urls, parent);
+    }
+
+    @Override
+    protected Class<?> loadClass(String name, boolean resolve) throws ClassNotFoundException {
+        // Always delegate java.*, javax.*, and our driver-api to parent
+        if (name.startsWith("java.")
+                || name.startsWith("javax.")
+                || name.startsWith("io.openmessaging.benchmark.driver.")) {
+            return super.loadClass(name, resolve);
+        }
+
+        // Child-first for everything else
+        synchronized (getClassLoadingLock(name)) {
+            Class<?> c = findLoadedClass(name);
+            if (c == null) {
+                try {
+                    c = findClass(name);
+                } catch (ClassNotFoundException e) {
+                    c = super.loadClass(name, resolve);
+                }
+            }
+            if (resolve) {
+                resolveClass(c);
+            }
+            return c;
+        }
+    }
+
+    /**
+     * Create an isolated classloader for the given driver home.
+     *
+     * @param driverName The driver name that will be loaded by the isolated classloader.
+     * @return The classloader for the driver.
+     * @throws Exception If the driver name is invalid.
+     */
+    public static ClassLoader forDriverFolder(File driverName) throws Exception {
+        List<URL> urls = new ArrayList<>();
+
+        if (driverName.isFile() && driverName.getName().endsWith(".jar")) {
+            // Single jar
+            urls.add(driverName.toURI().toURL());
+        } else if (driverName.isDirectory()) {
+            // Maven module layout: target/classes or a distribution folder
+            File classesDir = new File(driverName, "target/classes");
+            if (classesDir.exists()) {
+                urls.add(classesDir.toURI().toURL());
+            }
+
+            // Add all jars in root
+            File[] rootJars = driverName.listFiles((d, n) -> n.endsWith(".jar"));
+            if (rootJars != null) {
+                for (File f : rootJars) {
+                    urls.add(f.toURI().toURL());
+                }
+            }
+
+            // Add all jars in lib/
+            File libDir = new File(driverName, "lib");
+            if (libDir.exists()) {
+                File[] libJars = libDir.listFiles((d, n) -> n.endsWith(".jar"));
+                if (libJars != null) {
+                    for (File f : libJars) {
+                        urls.add(f.toURI().toURL());
+                    }
+                }
+            }
+        }
+
+        if (urls.isEmpty()) {
+            throw new IllegalArgumentException(
+                    "No jars found in driver home: " + driverName.getAbsolutePath());
+        }
+
+        log.info("Creating isolated classloader with {} URLs for {}", urls.size(), driverName);
+        return new IsolatedDriverLoader(
+                urls.toArray(new URL[0]), IsolatedDriverLoader.class.getClassLoader());
+    }
+
+    /**
+     * Reflectively instantiate a class from an isolated classloader.
+     *
+     * @param cl The classloader to use.
+     * @param className The class name to instantiate of the isolated benchmark driver.
+     * @param <T> The type of the class to instantiate.
+     * @return The instantiated class.
+     */
+    @SuppressWarnings("unchecked")
+    public static <T> T newInstance(ClassLoader cl, String className) throws Exception {
+        Class<?> cls = Class.forName(className, true, cl);
+        Constructor<?> ctor = cls.getDeclaredConstructor();
+        ctor.setAccessible(true);
+        return (T) ctor.newInstance();
+    }
+}

benchmark-framework/src/main/java/io/openmessaging/benchmark/utils/RandomGenerator.java

@@ -18,17 +18,35 @@
 public final class RandomGenerator {
 
     private static final Random random = new Random();
-    private static final String KAFKA_SAFE_CHARS =
-            "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-";
+    private static final char[] KAFKA_SAFE_CHARS = buildKafkaSafeChars();
 
     private RandomGenerator() {}
 
     public static String getRandomString() {
         // Generate a Kafka-safe random string (only alphanumerics, underscore, and hyphen)
         StringBuilder sb = new StringBuilder(7);
         for (int i = 0; i < 7; i++) {
-            sb.append(KAFKA_SAFE_CHARS.charAt(random.nextInt(KAFKA_SAFE_CHARS.length())));
+            sb.append(KAFKA_SAFE_CHARS[random.nextInt(KAFKA_SAFE_CHARS.length)]);
         }
         return sb.toString();
     }
+
+    private static char[] buildKafkaSafeChars() {
+        char[] chars = new char[26 + 26 + 10 + 2]; // a-z + A-Z + 0-9 + _-
+        int index = 0;
+
+        for (char c = 'a'; c <= 'z'; c++) {
+            chars[index++] = c;
+        }
+        for (char c = 'A'; c <= 'Z'; c++) {
+            chars[index++] = c;
+        }
+        for (char c = '0'; c <= '9'; c++) {
+            chars[index++] = c;
+        }
+        chars[index++] = '_';
+        chars[index] = '-';
+
+        return chars;
+    }
 }

benchmark-framework/src/main/java/io/openmessaging/benchmark/worker/BenchmarkWorker.java

@@ -25,7 +25,12 @@
 import org.slf4j.LoggerFactory;
 
 /** A benchmark worker that listen for tasks to perform. */
-public class BenchmarkWorker {
+public final class BenchmarkWorker {
+
+    private BenchmarkWorker() {
+        throw new UnsupportedOperationException(
+                "BenchmarkWorker is a CLI utility and cannot be instantiated");
+    }
 
     @SuppressWarnings("unused")
     static class Arguments {
@@ -44,6 +49,7 @@ static class Arguments {
         @Parameter(
                 names = {"-sp", "--stats-port"},
                 description = "Stats port to listen on")
+        @Deprecated(since = "0.0.3", forRemoval = true)
         private int statsPort = 8081;
 
         public boolean isHelp() {
@@ -59,7 +65,7 @@ public int getStatsPort() {
         }
     }
 
-    public static void main(String[] args) throws Exception {
+    static void main(String[] args) throws Exception {
         final Arguments arguments = new Arguments();
         JCommander jc = new JCommander(arguments);
         jc.setProgramName("benchmark-worker");

benchmark-framework/src/main/java/io/openmessaging/benchmark/worker/DistributedWorkersEnsemble.java

@@ -74,20 +74,19 @@ static int getNumberOfProducerWorkers(List<Worker> workers, boolean extraConsume
     }
 
     @Override
-    public void initializeDriver(File configurationFile) throws IOException {
+    public void initializeDriver(File configurationFile) throws Exception {
         workers.parallelStream()
                 .forEach(
                         w -> {
                             try {
                                 w.initializeDriver(configurationFile);
-                            } catch (IOException e) {
+                            } catch (Exception e) {
                                 throw new RuntimeException(e);
                             }
                         });
     }
 
     @Override
-    @SuppressWarnings("unchecked")
     public List<String> createTopics(TopicsInfo topicsInfo) throws IOException {
         return leader.createTopics(topicsInfo);
     }

benchmark-framework/src/main/java/io/openmessaging/benchmark/worker/LocalWorker.java

@@ -19,7 +19,6 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.ObjectWriter;
 import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
-import com.google.common.base.Preconditions;
 import io.micrometer.prometheusmetrics.PrometheusMeterRegistry;
 import io.netty.util.concurrent.DefaultThreadFactory;
 import io.openmessaging.benchmark.DriverConfiguration;
@@ -30,6 +29,7 @@
 import io.openmessaging.benchmark.driver.BenchmarkDriver.TopicInfo;
 import io.openmessaging.benchmark.driver.BenchmarkProducer;
 import io.openmessaging.benchmark.driver.ConsumerCallback;
+import io.openmessaging.benchmark.utils.IsolatedDriverLoader;
 import io.openmessaging.benchmark.utils.RandomGenerator;
 import io.openmessaging.benchmark.utils.Timer;
 import io.openmessaging.benchmark.utils.UniformRateLimiter;
@@ -79,19 +79,42 @@ public LocalWorker(PrometheusMeterRegistry statsLogger) {
     }
 
     @Override
-    public void initializeDriver(File driverConfigFile) throws IOException {
-        Preconditions.checkArgument(benchmarkDriver == null);
-        testCompleted = false;
+    public void initializeDriver(File driverConfigFile) throws Exception {
+        initializeDriver(driverConfigFile, null);
+    }
 
+    @Override
+    public void initializeDriver(File driverConfigFile, File isolatedDriverHome) throws Exception {
         DriverConfiguration driverConfiguration =
                 mapper.readValue(driverConfigFile, DriverConfiguration.class);
-
-        log.info("Driver: {}", writer.writeValueAsString(driverConfiguration));
+        log.info("Initializing driver: {}", writer.writeValueAsString(driverConfiguration));
 
         try {
-            benchmarkDriver =
-                    (BenchmarkDriver) Class.forName(driverConfiguration.driverClass).newInstance();
+            // Validate driver class is present
+            if (driverConfiguration.driverClass == null || driverConfiguration.driverClass.isEmpty()) {
+                throw new IllegalArgumentException("Driver configuration must specify a driverClass");
+            }
+
+            // Check if this driver should be loaded in isolation
+            boolean shouldIsolate =
+                    isolatedDriverHome != null && driverConfiguration.driverClass.contains("pravega");
+
+            if (shouldIsolate) {
+                log.info(
+                        "Loading driver {} with isolated classloader from {}",
+                        driverConfiguration.driverClass,
+                        isolatedDriverHome);
+                ClassLoader isolatedCL = IsolatedDriverLoader.forDriverFolder(isolatedDriverHome);
+                benchmarkDriver =
+                        IsolatedDriverLoader.newInstance(isolatedCL, driverConfiguration.driverClass);
+            } else {
+                // Standard loading for non-isolated drivers
+                benchmarkDriver =
+                        (BenchmarkDriver) Class.forName(driverConfiguration.driverClass).newInstance();
+            }
+
             benchmarkDriver.initialize(driverConfigFile, stats.getMeterRegistry());
+
         } catch (InstantiationException
                 | IllegalAccessException
                 | ClassNotFoundException

benchmark-framework/src/main/java/io/openmessaging/benchmark/worker/Worker.java

@@ -25,7 +25,12 @@
 
 public interface Worker extends AutoCloseable {
 
-    void initializeDriver(File configurationFile) throws IOException;
+    void initializeDriver(File driverConfigFile) throws Exception;
+
+    // New method to support isolated driver loading
+    default void initializeDriver(File driverConfigFile, File isolatedDriverHome) throws Exception {
+        initializeDriver(driverConfigFile);
+    }
 
     List<String> createTopics(TopicsInfo topicsInfo) throws IOException;
 

docker/Dockerfile

@@ -12,8 +12,7 @@
 # limitations under the License.
 #
 
-FROM eclipse-temurin:17
-
+FROM amazoncorretto:25-alpine
 ARG BENCHMARK_TARBALL
 
 # Using ADD instead of COPY because ${BENCHMARK_TARBALL} is an archive that needs to be extracted
@@ -24,8 +23,8 @@ RUN mv openmessaging-benchmark-* /benchmark
 WORKDIR /benchmark
 
 # Install vim
-ENV DEBIAN_FRONTEND=noninteractive
-RUN apt-get update && apt-get install -y vim tzdata && rm -rf /var/lib/apt/lists/*
+#ENV DEBIAN_FRONTEND=noninteractive
+#RUN apt-get update && apt-get install -y vim tzdata && rm -rf /var/lib/apt/lists/*
 
 # Start a shell by default
 CMD ["/bin/bash"]