Fix worker launch: add lib-common to the supervisor's worker classpath and restore the lib-worker drop-in directory

storm-dist/binary/final-package/src/main/assembly/common.xml

@@ -38,6 +38,12 @@
                 <include>*.jar</include>
             </includes>
         </fileSet>
+        <!-- lib-worker ships empty (README only): it stays on the worker classpath as the
+             drop-in point for worker-only jars, matching the pre-dedup 2.x layout. -->
+        <fileSet>
+            <directory>${project.basedir}/src/main/dist/lib-worker</directory>
+            <outputDirectory>lib-worker</outputDirectory>
+        </fileSet>
         <fileSet>
             <directory>${project.basedir}/../storm-webapp-bin/target/webapp/webapp/</directory>
             <outputDirectory>.</outputDirectory>

storm-dist/binary/final-package/src/main/dist/lib-worker/README.md

@@ -0,0 +1,15 @@
+# lib-worker
+
+Drop-in directory for worker-only jars.
+
+Jars placed here are added to the classpath of every worker JVM launched by
+the supervisor, but not to the daemon (nimbus / supervisor / ui) classpath.
+
+The distribution itself ships no worker-only jars: the jars shared by the
+daemons and the workers live in `lib-common/`, daemon-only jars in `lib/`.
+The classpaths are composed as:
+
+    daemon classpath = lib-common + lib
+    worker classpath = lib-common + lib-worker
+
+See also `extlib/`, which is added to both the daemon and worker classpaths.

storm-server/src/main/java/org/apache/storm/daemon/supervisor/BasicContainer.java

@@ -372,13 +372,18 @@ protected String getWildcardDir(File dir) {
     }
 
     protected List<String> frameworkClasspath(SimpleVersion topoVersion) {
+        // Jars shared by the daemon and worker classpaths are de-duplicated into lib-common
+        // (see storm-dist dedup-libs.py); storm-client and its dependencies live there, so the
+        // worker classpath needs lib-common in addition to lib-worker.
+        File stormCommonLibDir = new File(stormHome, "lib-common");
         File stormWorkerLibDir = new File(stormHome, "lib-worker");
         String topoConfDir = System.getenv("STORM_CONF_DIR") != null
                 ? System.getenv("STORM_CONF_DIR")
                 : new File(stormHome, "conf").getAbsolutePath();
         File stormExtlibDir = new File(stormHome, "extlib");
         String extcp = System.getenv("STORM_EXT_CLASSPATH");
         List<String> pathElements = new LinkedList<>();
+        pathElements.add(getWildcardDir(stormCommonLibDir));
         pathElements.add(getWildcardDir(stormWorkerLibDir));
         pathElements.add(getWildcardDir(stormExtlibDir));
         pathElements.add(extcp);

storm-server/src/test/java/org/apache/storm/daemon/supervisor/BasicContainerTest.java

@@ -31,11 +31,13 @@
 import org.apache.storm.utils.LocalState;
 import org.apache.storm.utils.SimpleVersion;
 import org.apache.storm.utils.Utils;
+import org.apache.storm.utils.VersionInfo;
 import org.junit.jupiter.api.Test;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
@@ -402,6 +404,47 @@ public void testLaunch() throws Exception {
                    "storm.log.dir", stormLogDir);
     }
 
+    @Test
+    public void testFrameworkClasspathIncludesSharedAndWorkerLibs() throws Exception {
+        final String topoId = "test_topology_classpath";
+        final int supervisorPort = 6628;
+        final int port = 8080;
+        final String stormHome = ContainerTest.asAbsPath("tmp", "storm-home");
+        final String stormLogDir = ContainerTest.asFile(".", "target").getCanonicalPath();
+        final String stormLocal = ContainerTest.asAbsPath("tmp", "storm-local");
+
+        final Map<String, Object> superConf = new HashMap<>();
+        superConf.put(Config.STORM_LOCAL_DIR, stormLocal);
+        superConf.put(Config.STORM_WORKERS_ARTIFACTS_DIR, stormLocal);
+
+        LocalAssignment la = new LocalAssignment();
+        la.set_topology_id(topoId);
+
+        AdvancedFSOps ops = mock(AdvancedFSOps.class);
+        when(ops.doRequiredTopoFilesExist(superConf, topoId)).thenReturn(true);
+
+        LocalState ls = mock(LocalState.class);
+        MockResourceIsolationManager iso = new MockResourceIsolationManager();
+
+        checkpoint(() -> {
+                       MockBasicContainer mc = new MockBasicContainer(ContainerType.LAUNCH, superConf,
+                           "SUPERVISOR", supervisorPort, port, la, iso, ls, "worker-id", new StormMetricsRegistry(),
+                           new HashMap<>(), ops, "profile");
+
+                       List<String> cp = mc.realFrameworkClasspath(VersionInfo.OUR_VERSION);
+
+                       // The distribution de-duplicates the jars shared by the daemon and worker
+                       // classpaths into lib-common; storm-client (LogWriter, Worker) ships there,
+                       // so a worker launched without lib-common on the classpath cannot start.
+                       String libCommon = stormHome + File.separator + "lib-common" + File.separator + "*";
+                       String libWorker = stormHome + File.separator + "lib-worker" + File.separator + "*";
+                       assertTrue(cp.contains(libCommon), "worker classpath must include lib-common/*, got: " + cp);
+                       assertTrue(cp.contains(libWorker), "worker classpath must include lib-worker/*, got: " + cp);
+                   },
+                   ConfigUtils.STORM_HOME, stormHome,
+                   "storm.log.dir", stormLogDir);
+    }
+
     @Test
     public void testLaunchStorm1version() throws Exception {
         final String topoId = "test_topology_storm_1.x";
@@ -712,6 +755,10 @@ protected List<String> frameworkClasspath(SimpleVersion version) {
             return Collections.singletonList("FRAMEWORK_CP");
         }
 
+        public List<String> realFrameworkClasspath(SimpleVersion version) {
+            return super.frameworkClasspath(version);
+        }
+
         @Override
         protected String javaLibraryPath(String stormRoot, Map<String, Object> conf) {
             return "JLP";