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";