Added in ability for users to include pre-installed jars to the classpath

storm-core/src/clj/backtype/storm/daemon/supervisor.clj

@@ -408,7 +408,7 @@
           stormroot (supervisor-stormdist-root conf storm-id)
           stormjar (supervisor-stormjar-path stormroot)
           storm-conf (read-supervisor-storm-conf conf storm-id)
-          classpath (add-to-classpath (current-classpath) [stormjar])
+          classpath (add-to-classpath (current-classpath) [stormjar (storm-conf TOPOLOGY-WORKER-CLASSPATH)])
           childopts (.replaceAll (str (conf WORKER-CHILDOPTS) " " (storm-conf TOPOLOGY-WORKER-CHILDOPTS))
                                  "%ID%"
                                  (str port))

storm-core/src/clj/backtype/storm/util.clj

@@ -477,7 +477,7 @@
   (System/getProperty "java.class.path"))
 
 (defn add-to-classpath [classpath paths]
-  (str/join ":" (cons classpath paths)))
+  (str/join ":" (cons classpath (remove str/blank? paths))))
 
 (defn ^ReentrantReadWriteLock mk-rw-lock []
   (ReentrantReadWriteLock.))

storm-core/src/jvm/backtype/storm/Config.java

@@ -641,6 +641,12 @@ public class Config extends HashMap<String, Object> {
     public static final String TOPOLOGY_WORKER_CHILDOPTS="topology.worker.childopts";
     public static final Object TOPOLOGY_WORKER_CHILDOPTS_SCHEMA = String.class;
 
+    /**
+     * Topology-specific options for the worker child process classpath.
+     */
+    public static final String TOPOLOGY_WORKER_CLASSPATH="topology.worker.classpath";
+    public static final Object TOPOLOGY_WORKER_CLASSPATH_SCHEMA = String.class;
+
     /**
      * This config is available for TransactionalSpouts, and contains the id ( a String) for
      * the transactional topology. This id is used to store the state of the transactional

storm-core/test/clj/backtype/storm/utils_test.clj

@@ -47,3 +47,11 @@
     ))
   )
 )
+
+(deftest test-add-to-classpath
+  (is (= (add-to-classpath "a:b", ["", nil]) "a:b"))
+  (is (= (add-to-classpath "a:b", ["c", nil]) "a:b:c"))
+  (is (= (add-to-classpath "a:b", ["c", "d:e"]) "a:b:c:d:e"))
+)
+
+