You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@apex.apache.org by chinmaykolhatkar <gi...@git.apache.org> on 2016/04/18 09:00:18 UTC
[GitHub] incubator-apex-core pull request: APEXCORE-304 Added support for l...
GitHub user chinmaykolhatkar opened a pull request:
https://github.com/apache/incubator-apex-core/pull/311
APEXCORE-304 Added support for libjars in local mode.
@PramodSSImmaneni Please review
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/chinmaykolhatkar/incubator-apex-core APEXCORE-304_LibJars
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-apex-core/pull/311.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #311
----
commit ea8a71faecd507cb4c12c133c9ccd2c1686d0a92
Author: chinmaykolhatkar <ch...@datatorrent.com>
Date: 2016-04-16T08:02:31Z
APEXCORE-304 Added support for libjars in local mode.
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-304 Added support for l...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/311#discussion_r60265837
--- Diff: api/src/main/java/com/datatorrent/api/StreamingApplication.java ---
@@ -47,6 +47,12 @@
String ENVIRONMENT = DT_PREFIX + "environment";
/**
+ * Use this config variable to to add any external library to classpath for the DAG.
+ * Value of this variable is same as StramAppLauncher.LIBJARS_CONF_KEY_NAME.
+ */
+ String LIBJARS_CONF_KEY_NAME = "tmplibjars";
--- End diff --
Hmmm.. got your point and agree with you.. I find attribute as a better place than properties for such configuration.
Here is my suggestion about exposing this as an attribute:
1. Have a attribute in DAGContext:
Attribute<Map<String, Boolean>> APP_EXTERNAL_JARS = new Attribute<Map<, >(new Map2String<String, Boolean>(",", "=", new String2String<>(), new String2String<>()));
2. We pick up this attribute in following places:
a. For cluster mode: StramAppLauncher.launchApp
b. For local mode (apex cli) : StramAppLauncher.runLocal
c. For local mode (ApplicationTest): LocalModeImpl.getController
3. For individual approaches, jar becomes available from that point onwards.
Can you share though on above approach?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-304 Added support for l...
Posted by tweise <gi...@git.apache.org>.
Github user tweise commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/311#discussion_r60105949
--- Diff: api/src/main/java/com/datatorrent/api/StreamingApplication.java ---
@@ -47,6 +47,12 @@
String ENVIRONMENT = DT_PREFIX + "environment";
/**
+ * Use this config variable to to add any external library to classpath for the DAG.
+ * Value of this variable is same as StramAppLauncher.LIBJARS_CONF_KEY_NAME.
+ */
+ String LIBJARS_CONF_KEY_NAME = "tmplibjars";
--- End diff --
I don't think this belongs into the API.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-304 Added support for l...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/311#discussion_r60265869
--- Diff: api/src/main/java/com/datatorrent/api/StreamingApplication.java ---
@@ -47,6 +47,12 @@
String ENVIRONMENT = DT_PREFIX + "environment";
/**
+ * Use this config variable to to add any external library to classpath for the DAG.
+ * Value of this variable is same as StramAppLauncher.LIBJARS_CONF_KEY_NAME.
+ */
+ String LIBJARS_CONF_KEY_NAME = "tmplibjars";
--- End diff --
Hmmm.. got your point and agree with you.. I find attribute as a better place than properties for such configuration.
Here is my suggestion about exposing this as an attribute:
1. Have a attribute in DAGContext:
Attribute<Map<String, Boolean>> APP_EXTERNAL_JARS = new Attribute<Map<, >(new Map2String<String, Boolean>(",", "=", new String2String<>(), new String2String<>()));
2. We pick up this attribute in following places:
a. For cluster mode: StramAppLauncher.launchApp
b. For local mode (apex cli) : StramAppLauncher.runLocal
c. For local mode (ApplicationTest): LocalModeImpl.getController
3. For individual approaches, jar becomes available from that point onwards.
Can you share though on above approach?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-304 Added support for l...
Posted by tweise <gi...@git.apache.org>.
Github user tweise commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/311#discussion_r61008036
--- Diff: engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java ---
@@ -2386,7 +2381,7 @@ public static void write(DAG dag, OutputStream os) throws IOException
public static LogicalPlan read(InputStream is) throws IOException, ClassNotFoundException
{
- return (LogicalPlan)new ObjectInputStream(is).readObject();
+ return (LogicalPlan)new ClassLoaderObjectInputStream(Thread.currentThread().getContextClassLoader(), is).readObject();
--- End diff --
Why is this required (or how was it working prior to this change)?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-304 Added support for l...
Posted by tweise <gi...@git.apache.org>.
Github user tweise commented on the pull request:
https://github.com/apache/incubator-apex-core/pull/311#issuecomment-211521200
Yes, the ticket should reflect the actual work being done.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-304 Added support for l...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar closed the pull request at:
https://github.com/apache/incubator-apex-core/pull/311
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-304 Added support for l...
Posted by tweise <gi...@git.apache.org>.
Github user tweise commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/311#discussion_r60498298
--- Diff: engine/src/test/java/com/datatorrent/stram/StramLocalClusterTest.java ---
@@ -274,4 +286,103 @@ public WindowGenerator setupWindowGenerator()
localCluster.shutdown();
}
+ @Test
+ public void testDynamicLoading() throws IOException, ClassNotFoundException
+ {
+ final String generatedJar = generatejar();
+ File file = new File(generatedJar);
+ final Class<?> pojo = URLClassLoader.newInstance(new URL[]{file.toURI().toURL()}).loadClass("POJO");
+
+ StreamingApplication app = new StreamingApplication()
+ {
+ @Override
+ public void populateDAG(DAG dag, Configuration conf)
+ {
+ TestGeneratorInputOperator genNode = dag.addOperator("genNode", TestGeneratorInputOperator.class);
+ genNode.setMaxTuples(2);
+
+ DynamicLoader dynamicLoader = dag.addOperator("DynamicLoader", new DynamicLoader());
+ dynamicLoader.setClass("POJO");
+
+ dag.addStream("fromNode1", genNode.outport, dynamicLoader.in);
+
+ dag.setInputPortAttribute(dynamicLoader.in, Context.PortContext.TUPLE_CLASS, pojo);
+ conf.set(LIBJARS_CONF_KEY_NAME, generatedJar);
+ }
+ };
+
+ LocalMode.runApp(app, 10000);
+
+ FileUtils.forceDelete(new File("src/test/resources/dynamicJar/POJO.class"));
--- End diff --
Why is the unit test (creating? and) deleting files in the source tree?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-304 Added support for l...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on the pull request:
https://github.com/apache/incubator-apex-core/pull/311#issuecomment-211519747
The idea behind this ticket is to allow user to provide external jars to classpath.
We already do have a way to provide external jars using tmplibjars conf property.
But this works only for cluster mode and not for local mode.
This change change makes local mode also consume tmplibjars conf variable.
This way the purpose of served without creating extra APIs etc...
I can change the description of the ticket is that's right to do.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-304 Added support for l...
Posted by tweise <gi...@git.apache.org>.
Github user tweise commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/311#discussion_r60498165
--- Diff: engine/src/test/java/com/datatorrent/stram/StramLocalClusterTest.java ---
@@ -274,4 +286,103 @@ public WindowGenerator setupWindowGenerator()
localCluster.shutdown();
}
+ @Test
+ public void testDynamicLoading() throws IOException, ClassNotFoundException
+ {
+ final String generatedJar = generatejar();
+ File file = new File(generatedJar);
+ final Class<?> pojo = URLClassLoader.newInstance(new URL[]{file.toURI().toURL()}).loadClass("POJO");
+
+ StreamingApplication app = new StreamingApplication()
+ {
+ @Override
+ public void populateDAG(DAG dag, Configuration conf)
+ {
+ TestGeneratorInputOperator genNode = dag.addOperator("genNode", TestGeneratorInputOperator.class);
+ genNode.setMaxTuples(2);
+
+ DynamicLoader dynamicLoader = dag.addOperator("DynamicLoader", new DynamicLoader());
+ dynamicLoader.setClass("POJO");
+
+ dag.addStream("fromNode1", genNode.outport, dynamicLoader.in);
+
+ dag.setInputPortAttribute(dynamicLoader.in, Context.PortContext.TUPLE_CLASS, pojo);
+ conf.set(LIBJARS_CONF_KEY_NAME, generatedJar);
+ }
+ };
+
+ LocalMode.runApp(app, 10000);
--- End diff --
Please structure the unit tests so that you wait for expected result instead of waiting for arbitrary time. There are examples in Malhar for this.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-304 Added support for l...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
GitHub user chinmaykolhatkar reopened a pull request:
https://github.com/apache/incubator-apex-core/pull/311
APEXCORE-304 Added support for libjars in local mode.
@PramodSSImmaneni Please review
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/chinmaykolhatkar/incubator-apex-core APEXCORE-304_LibJars
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-apex-core/pull/311.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #311
----
commit f5c1efb13d4dfb8b59e1ffaf134478cd9f5a5109
Author: chinmaykolhatkar <ch...@datatorrent.com>
Date: 2016-04-21T10:09:49Z
APEXCORE-304 Moved LogicalPlan.LIBRARY_JARS to DAGContext.
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-304 Added support for l...
Posted by tweise <gi...@git.apache.org>.
Github user tweise commented on the pull request:
https://github.com/apache/incubator-apex-core/pull/311#issuecomment-211509289
The ticket seems to suggest a broader change than just local mode. What am I missing?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-304 Added support for l...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on the pull request:
https://github.com/apache/incubator-apex-core/pull/311#issuecomment-213531795
All comments are taken care of.
@tweise @PramodSSImmaneni Please review.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-304 Added support for l...
Posted by tweise <gi...@git.apache.org>.
Github user tweise commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/311#discussion_r61008300
--- Diff: engine/src/test/java/com/datatorrent/stram/StramLocalClusterTest.java ---
@@ -274,4 +287,113 @@ public WindowGenerator setupWindowGenerator()
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 = "target/";
--- End diff --
use StramTestSupport.TestMeta
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-304 Added support for l...
Posted by tweise <gi...@git.apache.org>.
Github user tweise commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/311#discussion_r60498799
--- Diff: api/src/main/java/com/datatorrent/api/StreamingApplication.java ---
@@ -47,6 +47,12 @@
String ENVIRONMENT = DT_PREFIX + "environment";
/**
+ * Use this config variable to to add any external library to classpath for the DAG.
+ * Value of this variable is same as StramAppLauncher.LIBJARS_CONF_KEY_NAME.
+ */
+ String LIBJARS_CONF_KEY_NAME = "tmplibjars";
--- End diff --
There already is an attribute that was taken out of the public API some time ago... LogicalPlan.LIBRARY_JARS
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-304 Added support for l...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/311#discussion_r61535209
--- Diff: engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java ---
@@ -2386,7 +2381,7 @@ public static void write(DAG dag, OutputStream os) throws IOException
public static LogicalPlan read(InputStream is) throws IOException, ClassNotFoundException
{
- return (LogicalPlan)new ObjectInputStream(is).readObject();
+ return (LogicalPlan)new ClassLoaderObjectInputStream(Thread.currentThread().getContextClassLoader(), is).readObject();
--- End diff --
This is required for Local Mode case.
Lets say a jar path is set to LIBRARY_JARS and this jar contains a POJO that needs to be set to TUPLE_CLASS attr. Now, one can do that in populateDAG method, that would cause a problem in LocalMode as follows:
1. populateDAG sets TUPLE_CLASS attr with new POJO class - Works.
2. Local mode serializes the LogicalPlan - Works.
3. Local mode deserialized the LogicalPlan - Fails.
3rd step fails becuase The ObjectInputStream deserializer uses System ClassLoader which does that have that POJO. ClassLoaderObjectInputStream is an overridden version of ObjectInputStream which allows to set the custom classloader.
By this time of callstack, Thread.currentThread().getClassLoader() will have a classloader which is super of all.
As previously, there was no functionality of LIBRARY_JARS, I guess this was not tested and so did not surface as an issue.
For cluster mode, there is no difference. i.e. There is only 1 classloader i.e. System.Classloader and that has new jars as well.
i.e. For cluster mode, Thread.currentThread().getClassLoader() = System classloader
i.e. ClassLoaderObjectInputStream will behave as if ObjectInputStream.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-304 Added support for l...
Posted by tweise <gi...@git.apache.org>.
Github user tweise commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/311#discussion_r60113016
--- Diff: api/src/main/java/com/datatorrent/api/StreamingApplication.java ---
@@ -47,6 +47,12 @@
String ENVIRONMENT = DT_PREFIX + "environment";
/**
+ * Use this config variable to to add any external library to classpath for the DAG.
+ * Value of this variable is same as StramAppLauncher.LIBJARS_CONF_KEY_NAME.
+ */
+ String LIBJARS_CONF_KEY_NAME = "tmplibjars";
--- End diff --
Just looking at the string literal suggests something isn't right here. We configure applications through attributes, not random properties in the config file. Can you please check if there is a better way to express this dependency.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-304 Added support for l...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/311#discussion_r61534824
--- Diff: engine/src/test/java/com/datatorrent/stram/StramLocalClusterTest.java ---
@@ -274,4 +287,113 @@ public WindowGenerator setupWindowGenerator()
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 = "target/";
--- End diff --
Done.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-304 Added support for l...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/incubator-apex-core/pull/311
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-304 Added support for l...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on the pull request:
https://github.com/apache/incubator-apex-core/pull/311#issuecomment-211265259
Not sure why the travis build is failing.. Travis command is passing successfully when run locally.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-304 Added support for l...
Posted by tweise <gi...@git.apache.org>.
Github user tweise commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/311#discussion_r60498409
--- Diff: engine/src/test/java/com/datatorrent/stram/StramLocalClusterTest.java ---
@@ -274,4 +286,103 @@ public WindowGenerator setupWindowGenerator()
localCluster.shutdown();
}
+ @Test
+ public void testDynamicLoading() throws IOException, ClassNotFoundException
+ {
+ final String generatedJar = generatejar();
+ File file = new File(generatedJar);
+ final Class<?> pojo = URLClassLoader.newInstance(new URL[]{file.toURI().toURL()}).loadClass("POJO");
+
+ StreamingApplication app = new StreamingApplication()
+ {
+ @Override
+ public void populateDAG(DAG dag, Configuration conf)
+ {
+ TestGeneratorInputOperator genNode = dag.addOperator("genNode", TestGeneratorInputOperator.class);
+ genNode.setMaxTuples(2);
+
+ DynamicLoader dynamicLoader = dag.addOperator("DynamicLoader", new DynamicLoader());
+ dynamicLoader.setClass("POJO");
+
+ dag.addStream("fromNode1", genNode.outport, dynamicLoader.in);
+
+ dag.setInputPortAttribute(dynamicLoader.in, Context.PortContext.TUPLE_CLASS, pojo);
+ conf.set(LIBJARS_CONF_KEY_NAME, generatedJar);
+ }
+ };
+
+ LocalMode.runApp(app, 10000);
+
+ FileUtils.forceDelete(new File("src/test/resources/dynamicJar/POJO.class"));
+ FileUtils.forceDelete(new File("src/test/resources/dynamicJar/testPOJO.jar"));
+ }
+
+ public static class DynamicLoader extends BaseOperator
+ {
+ private String classToLoad;
+
+ public final transient DefaultInputPort in = new DefaultInputPort()
+ {
+ @Override
+ public void setup(Context.PortContext context)
+ {
+ Class<?> value = context.getValue(Context.PortContext.TUPLE_CLASS);
+
+ if (!value.getName().equals("POJO")) {
+ throw new RuntimeException("Class name not matching. Name is " + value.getName());
+ }
+ }
+
+ @Override
+ public void process(Object tuple)
+ {
+
+ }
+ };
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ try {
+ Thread.currentThread().getContextClassLoader().loadClass(classToLoad);
+ Assert.assertTrue(true);
+ } catch (ClassNotFoundException e) {
+ Assert.fail(e.getMessage());
--- End diff --
propagate exception
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-304 Added support for l...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/311#discussion_r60111129
--- Diff: api/src/main/java/com/datatorrent/api/StreamingApplication.java ---
@@ -47,6 +47,12 @@
String ENVIRONMENT = DT_PREFIX + "environment";
/**
+ * Use this config variable to to add any external library to classpath for the DAG.
+ * Value of this variable is same as StramAppLauncher.LIBJARS_CONF_KEY_NAME.
+ */
+ String LIBJARS_CONF_KEY_NAME = "tmplibjars";
--- End diff --
The reason why it's added here is constant for tmplibjars is present in StramLauncher which is part of apex-engine and if this variable is to be used in malhar, apex-engine won't be added as dependency to malhar.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---