diff --git a/storm-dist/binary/final-package/src/main/assembly/common.xml b/storm-dist/binary/final-package/src/main/assembly/common.xml
index 6b50cc8ada8..402fd3837b2 100644
--- a/storm-dist/binary/final-package/src/main/assembly/common.xml
+++ b/storm-dist/binary/final-package/src/main/assembly/common.xml
@@ -38,6 +38,12 @@
*.jar
+
+
+ ${project.basedir}/src/main/dist/lib-worker
+ lib-worker
+
${project.basedir}/../storm-webapp-bin/target/webapp/webapp/
.
diff --git a/storm-dist/binary/final-package/src/main/dist/lib-worker/README.md b/storm-dist/binary/final-package/src/main/dist/lib-worker/README.md
new file mode 100644
index 00000000000..65f2d6a5a9a
--- /dev/null
+++ b/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.
diff --git a/storm-server/src/main/java/org/apache/storm/daemon/supervisor/BasicContainer.java b/storm-server/src/main/java/org/apache/storm/daemon/supervisor/BasicContainer.java
index 7583a96db79..35ce6c56f7e 100644
--- a/storm-server/src/main/java/org/apache/storm/daemon/supervisor/BasicContainer.java
+++ b/storm-server/src/main/java/org/apache/storm/daemon/supervisor/BasicContainer.java
@@ -372,6 +372,10 @@ protected String getWildcardDir(File dir) {
}
protected List 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")
@@ -379,6 +383,7 @@ protected List frameworkClasspath(SimpleVersion topoVersion) {
File stormExtlibDir = new File(stormHome, "extlib");
String extcp = System.getenv("STORM_EXT_CLASSPATH");
List pathElements = new LinkedList<>();
+ pathElements.add(getWildcardDir(stormCommonLibDir));
pathElements.add(getWildcardDir(stormWorkerLibDir));
pathElements.add(getWildcardDir(stormExtlibDir));
pathElements.add(extcp);
diff --git a/storm-server/src/test/java/org/apache/storm/daemon/supervisor/BasicContainerTest.java b/storm-server/src/test/java/org/apache/storm/daemon/supervisor/BasicContainerTest.java
index a4f6ec93aa1..4f941d42ef4 100644
--- a/storm-server/src/test/java/org/apache/storm/daemon/supervisor/BasicContainerTest.java
+++ b/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 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 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 frameworkClasspath(SimpleVersion version) {
return Collections.singletonList("FRAMEWORK_CP");
}
+ public List realFrameworkClasspath(SimpleVersion version) {
+ return super.frameworkClasspath(version);
+ }
+
@Override
protected String javaLibraryPath(String stormRoot, Map conf) {
return "JLP";