You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@apex.apache.org by th...@apache.org on 2016/04/29 07:38:36 UTC
[1/2] incubator-apex-core git commit: APEXCORE-304 Add support for
external jars to be added to DAG as dependency.
Repository: incubator-apex-core
Updated Branches:
refs/heads/master 20607b59a -> 37e4c74ec
APEXCORE-304 Add support for external jars to be added to DAG as dependency.
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/f71bb0ed
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/f71bb0ed
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/f71bb0ed
Branch: refs/heads/master
Commit: f71bb0edfd5e10fdf55343a151e0a41e2c4cba7a
Parents: c2008f2
Author: chinmaykolhatkar <ch...@datatorrent.com>
Authored: Thu Apr 21 15:39:49 2016 +0530
Committer: chinmaykolhatkar <ch...@datatorrent.com>
Committed: Fri Apr 29 10:48:24 2016 +0530
----------------------------------------------------------------------
.../main/java/com/datatorrent/api/Context.java | 8 ++
.../stram/LaunchContainerRunnable.java | 2 +-
.../com/datatorrent/stram/LocalModeImpl.java | 29 +++++
.../java/com/datatorrent/stram/StramClient.java | 5 +-
.../stram/StreamingContainerManager.java | 2 +-
.../stram/client/StramAppLauncher.java | 18 ++-
.../stram/plan/logical/LogicalPlan.java | 11 +-
.../stram/StramLocalClusterTest.java | 122 +++++++++++++++++++
engine/src/test/resources/dynamicJar/POJO.java | 49 ++++++++
9 files changed, 231 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/f71bb0ed/api/src/main/java/com/datatorrent/api/Context.java
----------------------------------------------------------------------
diff --git a/api/src/main/java/com/datatorrent/api/Context.java b/api/src/main/java/com/datatorrent/api/Context.java
index 7e19a56..a0f3ad3 100644
--- a/api/src/main/java/com/datatorrent/api/Context.java
+++ b/api/src/main/java/com/datatorrent/api/Context.java
@@ -519,6 +519,14 @@ public interface Context
*/
Attribute<AffinityRulesSet> AFFINITY_RULES_SET = new Attribute<AffinityRulesSet>(new JsonStringCodec<AffinityRulesSet>(AffinityRulesSet.class));
+ /**
+ * Comma separated list of jar file dependencies to be deployed with the application.
+ * The launcher will combine the list with built-in dependencies and those specified
+ * that are made available through the distributed file system to application master
+ * and child containers.
+ */
+ Attribute<String> LIBRARY_JARS = new Attribute<>(new StringCodec.String2String());
+
@SuppressWarnings(value = "FieldNameHidesFieldInSuperclass")
long serialVersionUID = AttributeMap.AttributeInitializer.initialize(DAGContext.class);
}
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/f71bb0ed/engine/src/main/java/com/datatorrent/stram/LaunchContainerRunnable.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/LaunchContainerRunnable.java b/engine/src/main/java/com/datatorrent/stram/LaunchContainerRunnable.java
index 8a8db79..e9dd72b 100644
--- a/engine/src/main/java/com/datatorrent/stram/LaunchContainerRunnable.java
+++ b/engine/src/main/java/com/datatorrent/stram/LaunchContainerRunnable.java
@@ -168,7 +168,7 @@ public class LaunchContainerRunnable implements Runnable
try {
// child VM dependencies
try (FileSystem fs = StramClientUtils.newFileSystemInstance(nmClient.getConfig())) {
- addFilesToLocalResources(LocalResourceType.FILE, dag.getAttributes().get(LogicalPlan.LIBRARY_JARS), localResources, fs);
+ addFilesToLocalResources(LocalResourceType.FILE, dag.getAttributes().get(Context.DAGContext.LIBRARY_JARS), localResources, fs);
String archives = dag.getAttributes().get(LogicalPlan.ARCHIVES);
if (archives != null) {
addFilesToLocalResources(LocalResourceType.ARCHIVE, archives, localResources, fs);
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/f71bb0ed/engine/src/main/java/com/datatorrent/stram/LocalModeImpl.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/LocalModeImpl.java b/engine/src/main/java/com/datatorrent/stram/LocalModeImpl.java
index e51af6b..3fedc7c 100644
--- a/engine/src/main/java/com/datatorrent/stram/LocalModeImpl.java
+++ b/engine/src/main/java/com/datatorrent/stram/LocalModeImpl.java
@@ -18,8 +18,14 @@
*/
package com.datatorrent.stram;
+import java.io.File;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.net.URLClassLoader;
+
import org.apache.hadoop.conf.Configuration;
+import com.datatorrent.api.Context;
import com.datatorrent.api.DAG;
import com.datatorrent.api.LocalMode;
import com.datatorrent.api.StreamingApplication;
@@ -66,9 +72,32 @@ public class LocalModeImpl extends LocalMode
public Controller getController()
{
try {
+ addLibraryJarsToClasspath(lp);
return new StramLocalCluster(lp);
} catch (Exception e) {
throw new RuntimeException("Error creating local cluster", e);
}
}
+
+ private void addLibraryJarsToClasspath(LogicalPlan lp) throws MalformedURLException
+ {
+ String libJarsCsv = lp.getAttributes().get(Context.DAGContext.LIBRARY_JARS);
+
+ if (libJarsCsv != null && libJarsCsv.length() != 0) {
+ String[] split = libJarsCsv.split(StramClient.LIB_JARS_SEP);
+ if (split.length != 0) {
+ URL[] urlList = new URL[split.length];
+ for (int i = 0; i < split.length; i++) {
+ File file = new File(split[i]);
+ urlList[i] = file.toURI().toURL();
+ }
+
+ // Set class loader.
+ ClassLoader prevCl = Thread.currentThread().getContextClassLoader();
+ URLClassLoader cl = URLClassLoader.newInstance(urlList, prevCl);
+ Thread.currentThread().setContextClassLoader(cl);
+ }
+ }
+
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/f71bb0ed/engine/src/main/java/com/datatorrent/stram/StramClient.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/StramClient.java b/engine/src/main/java/com/datatorrent/stram/StramClient.java
index c1dfffd..daca67e 100644
--- a/engine/src/main/java/com/datatorrent/stram/StramClient.java
+++ b/engine/src/main/java/com/datatorrent/stram/StramClient.java
@@ -74,6 +74,7 @@ import org.apache.log4j.DTLoggerFactory;
import com.google.common.base.Objects;
import com.google.common.collect.Lists;
+import com.datatorrent.api.Context;
import com.datatorrent.api.Context.OperatorContext;
import com.datatorrent.api.StorageAgent;
import com.datatorrent.common.util.AsyncFSStorageAgent;
@@ -223,7 +224,7 @@ public class StramClient
localJarFiles.add(jar);
}
- String libJarsPath = dag.getValue(LogicalPlan.LIBRARY_JARS);
+ String libJarsPath = dag.getValue(Context.DAGContext.LIBRARY_JARS);
if (!StringUtils.isEmpty(libJarsPath)) {
String[] libJars = StringUtils.splitByWholeSeparator(libJarsPath, LIB_JARS_SEP);
localJarFiles.addAll(Arrays.asList(libJars));
@@ -442,7 +443,7 @@ public class StramClient
String libJarsCsv = copyFromLocal(fs, appPath, localJarFiles.toArray(new String[]{}));
LOG.info("libjars: {}", libJarsCsv);
- dag.getAttributes().put(LogicalPlan.LIBRARY_JARS, libJarsCsv);
+ dag.getAttributes().put(Context.DAGContext.LIBRARY_JARS, libJarsCsv);
LaunchContainerRunnable.addFilesToLocalResources(LocalResourceType.FILE, libJarsCsv, localResources, fs);
if (archives != null) {
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/f71bb0ed/engine/src/main/java/com/datatorrent/stram/StreamingContainerManager.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/StreamingContainerManager.java b/engine/src/main/java/com/datatorrent/stram/StreamingContainerManager.java
index 6a8ee9c..37f63b2 100644
--- a/engine/src/main/java/com/datatorrent/stram/StreamingContainerManager.java
+++ b/engine/src/main/java/com/datatorrent/stram/StreamingContainerManager.java
@@ -3131,7 +3131,7 @@ public class StreamingContainerManager implements PlanContext
lp.setAttribute(LogicalPlan.APPLICATION_ID, appId);
lp.setAttribute(LogicalPlan.APPLICATION_PATH, newApp.assertAppPath());
- lp.setAttribute(LogicalPlan.LIBRARY_JARS, newApp.getValue(LogicalPlan.LIBRARY_JARS));
+ lp.setAttribute(Context.DAGContext.LIBRARY_JARS, newApp.getValue(Context.DAGContext.LIBRARY_JARS));
lp.setAttribute(LogicalPlan.ARCHIVES, newApp.getValue(LogicalPlan.ARCHIVES));
this.finals = new FinalVars(finals, lp);
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/f71bb0ed/engine/src/main/java/com/datatorrent/stram/client/StramAppLauncher.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/client/StramAppLauncher.java b/engine/src/main/java/com/datatorrent/stram/client/StramAppLauncher.java
index a1197a5..bd58e35 100644
--- a/engine/src/main/java/com/datatorrent/stram/client/StramAppLauncher.java
+++ b/engine/src/main/java/com/datatorrent/stram/client/StramAppLauncher.java
@@ -285,7 +285,7 @@ public class StramAppLauncher
// don't rely on object deserialization for changing the app id in the future.
try {
JSONObject attributes = metaJson.getJSONObject("attributes");
- originalLibJars = attributes.getString(LogicalPlan.LIBRARY_JARS.getSimpleName());
+ originalLibJars = attributes.getString(Context.DAGContext.LIBRARY_JARS.getSimpleName());
recoveryAppName = attributes.getString(Context.DAGContext.APPLICATION_NAME.getSimpleName());
} catch (JSONException ex) {
recoveryAppName = "Recovery App From " + originalAppId;
@@ -532,10 +532,22 @@ public class StramAppLauncher
*/
public void runLocal(AppFactory appConfig) throws Exception
{
+ propertiesBuilder.conf.setEnum(StreamingApplication.ENVIRONMENT, StreamingApplication.Environment.LOCAL);
+ LogicalPlan lp = appConfig.createApp(propertiesBuilder);
+
+ String libJarsCsv = lp.getAttributes().get(Context.DAGContext.LIBRARY_JARS);
+ if (libJarsCsv != null && libJarsCsv.length() != 0) {
+ String[] split = libJarsCsv.split(StramClient.LIB_JARS_SEP);
+ for (String jarPath : split) {
+ File file = new File(jarPath);
+ URL url = file.toURI().toURL();
+ launchDependencies.add(url);
+ }
+ }
+
// local mode requires custom classes to be resolved through the context class loader
loadDependencies();
- propertiesBuilder.conf.setEnum(StreamingApplication.ENVIRONMENT, StreamingApplication.Environment.LOCAL);
- StramLocalCluster lc = new StramLocalCluster(appConfig.createApp(propertiesBuilder));
+ StramLocalCluster lc = new StramLocalCluster(lp);
lc.run();
}
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/f71bb0ed/engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java b/engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java
index e3f6987..15969b7 100644
--- a/engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java
+++ b/engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java
@@ -61,6 +61,8 @@ import javax.validation.constraints.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+
+import org.apache.commons.io.input.ClassLoaderObjectInputStream;
import org.apache.commons.lang.builder.ToStringBuilder;
import org.apache.commons.lang.builder.ToStringStyle;
import org.apache.hadoop.conf.Configuration;
@@ -158,13 +160,6 @@ public class LogicalPlan implements Serializable, DAG
public static Attribute<String> KEY_TAB_FILE = new Attribute<>((String)null, new StringCodec.String2String());
public static Attribute<Double> TOKEN_REFRESH_ANTICIPATORY_FACTOR = new Attribute<>(0.7);
/**
- * Comma separated list of jar file dependencies to be deployed with the application.
- * The launcher will combine the list with built-in dependencies and those specified
- * that are made available through the distributed file system to application master
- * and child containers.
- */
- public static Attribute<String> LIBRARY_JARS = new Attribute<>(new StringCodec.String2String());
- /**
* Comma separated list of archives to be deployed with the application.
* The launcher will include the archives into the final set of resources
* that are made available through the distributed file system to application master
@@ -2386,7 +2381,7 @@ public class LogicalPlan implements Serializable, DAG
public static LogicalPlan read(InputStream is) throws IOException, ClassNotFoundException
{
- return (LogicalPlan)new ObjectInputStream(is).readObject();
+ return (LogicalPlan)new ClassLoaderObjectInputStream(Thread.currentThread().getContextClassLoader(), is).readObject();
}
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/f71bb0ed/engine/src/test/java/com/datatorrent/stram/StramLocalClusterTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/StramLocalClusterTest.java b/engine/src/test/java/com/datatorrent/stram/StramLocalClusterTest.java
index e620141..1a5046c 100644
--- a/engine/src/test/java/com/datatorrent/stram/StramLocalClusterTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/StramLocalClusterTest.java
@@ -22,8 +22,11 @@ import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.io.LineNumberReader;
+import java.net.URL;
+import java.net.URLClassLoader;
import java.util.Arrays;
import java.util.Map;
+import java.util.concurrent.CountDownLatch;
import org.junit.After;
import org.junit.Assert;
@@ -33,7 +36,16 @@ import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.commons.io.IOUtils;
+import org.apache.hadoop.conf.Configuration;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DAG;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.LocalMode;
+import com.datatorrent.api.StreamingApplication;
import com.datatorrent.common.util.AsyncFSStorageAgent;
+import com.datatorrent.common.util.BaseOperator;
import com.datatorrent.stram.StramLocalCluster.LocalStreamingContainer;
import com.datatorrent.stram.StramLocalCluster.MockComponentFactory;
import com.datatorrent.stram.api.Checkpoint;
@@ -48,6 +60,7 @@ import com.datatorrent.stram.plan.physical.PTOperator;
import com.datatorrent.stram.support.ManualScheduledExecutorService;
import com.datatorrent.stram.support.StramTestSupport;
+
public class StramLocalClusterTest
{
private static final Logger LOG = LoggerFactory.getLogger(StramLocalClusterTest.class);
@@ -274,4 +287,113 @@ public class StramLocalClusterTest
localCluster.shutdown();
}
+ @Test
+ public void testDynamicLoading() throws Exception
+ {
+ String generatedJar = generatejar("POJO");
+ URLClassLoader uCl = URLClassLoader.newInstance(new URL[] {new File(generatedJar).toURI().toURL()});
+ Class<?> pojo = uCl.loadClass("POJO");
+
+ DynamicLoaderApp app = new DynamicLoaderApp();
+ app.generatedJar = generatedJar;
+ app.pojo = pojo;
+
+ LocalMode lma = LocalMode.newInstance();
+ lma.prepareDAG(app, new Configuration());
+ LocalMode.Controller lc = lma.getController();
+ lc.runAsync();
+ DynamicLoaderApp.latch.await();
+ Assert.assertTrue(DynamicLoaderApp.passed);
+ lc.shutdown();
+ }
+
+ static class DynamicLoaderApp implements StreamingApplication
+ {
+ static boolean passed = false;
+ static CountDownLatch latch = new CountDownLatch(2);
+
+ DynamicLoader test;
+ String generatedJar;
+ Class<?> pojo;
+
+ @Override
+ public void populateDAG(DAG dag, Configuration conf)
+ {
+ TestGeneratorInputOperator input = dag.addOperator("Input", new TestGeneratorInputOperator());
+ test = dag.addOperator("Test", new DynamicLoader());
+
+ dag.addStream("S1", input.outport, test.input);
+ dag.setAttribute(Context.DAGContext.LIBRARY_JARS, generatedJar);
+ dag.setInputPortAttribute(test.input, Context.PortContext.TUPLE_CLASS, pojo);
+ }
+ }
+
+ static class DynamicLoader extends BaseOperator
+ {
+ public final transient DefaultInputPort input = new DefaultInputPort()
+ {
+ @Override
+ public void setup(Context.PortContext context)
+ {
+ Class<?> value = context.getValue(Context.PortContext.TUPLE_CLASS);
+ if (value.getName().equals("POJO")) {
+ DynamicLoaderApp.passed = true;
+ } else {
+ DynamicLoaderApp.passed = false;
+ }
+ DynamicLoaderApp.latch.countDown();
+ }
+
+ @Override
+ public void process(Object tuple)
+ {
+ }
+ };
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ ClassLoader cl = Thread.currentThread().getContextClassLoader();
+ try {
+ cl.loadClass("POJO");
+ } catch (ClassNotFoundException e) {
+ DynamicLoaderApp.passed = false;
+ DynamicLoaderApp.latch.countDown();
+ throw new RuntimeException(e);
+ }
+
+ try {
+ Class.forName("POJO", true, Thread.currentThread().getContextClassLoader());
+ } catch (ClassNotFoundException e) {
+ DynamicLoaderApp.passed = false;
+ DynamicLoaderApp.latch.countDown();
+ throw new RuntimeException(e);
+ }
+
+ DynamicLoaderApp.passed = true;
+ DynamicLoaderApp.latch.countDown();
+ }
+ }
+
+ private String generatejar(String pojoClassName) throws IOException, InterruptedException
+ {
+ String sourceDir = "src/test/resources/dynamicJar/";
+ String destDir = testMeta.getPath();
+
+ Process p = Runtime.getRuntime()
+ .exec(new String[] {"javac", "-d", destDir, sourceDir + pojoClassName + ".java"}, null, null);
+ IOUtils.copy(p.getInputStream(), System.out);
+ IOUtils.copy(p.getErrorStream(), System.err);
+ Assert.assertEquals(0, p.waitFor());
+
+ p = Runtime.getRuntime()
+ .exec(new String[] {"jar", "-cf", pojoClassName + ".jar", pojoClassName + ".class"}, null, new File(destDir));
+ IOUtils.copy(p.getInputStream(), System.out);
+ IOUtils.copy(p.getErrorStream(), System.err);
+ Assert.assertEquals(0, p.waitFor());
+
+ return new File(destDir, pojoClassName + ".jar").getAbsolutePath();
+ }
+
+
}
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/f71bb0ed/engine/src/test/resources/dynamicJar/POJO.java
----------------------------------------------------------------------
diff --git a/engine/src/test/resources/dynamicJar/POJO.java b/engine/src/test/resources/dynamicJar/POJO.java
new file mode 100644
index 0000000..8868544
--- /dev/null
+++ b/engine/src/test/resources/dynamicJar/POJO.java
@@ -0,0 +1,49 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+/**
+ * This file is not directly used anywhere. This will be compiled and packages as jar by unit test to verify
+ * dynamic loading.
+ */
+public class POJO
+{
+ private int a;
+ private String b;
+
+ public int getA()
+ {
+ return a;
+ }
+
+ public void setA(int a)
+ {
+ this.a = a;
+ }
+
+ public String getB()
+ {
+ return b;
+ }
+
+ public void setB(String b)
+ {
+ this.b = b;
+ }
+}
+
[2/2] incubator-apex-core git commit: Merge branch
'APEXCORE-304_LibJars' of
https://github.com/chinmaykolhatkar/incubator-apex-core
Posted by th...@apache.org.
Merge branch 'APEXCORE-304_LibJars' of https://github.com/chinmaykolhatkar/incubator-apex-core
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/37e4c74e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/37e4c74e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/37e4c74e
Branch: refs/heads/master
Commit: 37e4c74ecc2b89fba69b42f01ceead3ec2fe5734
Parents: 20607b5 f71bb0e
Author: Thomas Weise <th...@datatorrent.com>
Authored: Thu Apr 28 22:37:12 2016 -0700
Committer: Thomas Weise <th...@datatorrent.com>
Committed: Thu Apr 28 22:37:12 2016 -0700
----------------------------------------------------------------------
.../main/java/com/datatorrent/api/Context.java | 8 ++
.../stram/LaunchContainerRunnable.java | 2 +-
.../com/datatorrent/stram/LocalModeImpl.java | 29 +++++
.../java/com/datatorrent/stram/StramClient.java | 5 +-
.../stram/StreamingContainerManager.java | 2 +-
.../stram/client/StramAppLauncher.java | 18 ++-
.../stram/plan/logical/LogicalPlan.java | 11 +-
.../stram/StramLocalClusterTest.java | 122 +++++++++++++++++++
engine/src/test/resources/dynamicJar/POJO.java | 49 ++++++++
9 files changed, 231 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/37e4c74e/engine/src/main/java/com/datatorrent/stram/StramClient.java
----------------------------------------------------------------------