beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From k...@apache.org
Subject [13/36] beam git commit: mr-runner: add JarClassInstanceFactory to run ValidatesRunner tests.
Date Thu, 07 Sep 2017 18:39:22 GMT
mr-runner: add JarClassInstanceFactory to run ValidatesRunner tests.


Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/ee1cce92
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/ee1cce92
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/ee1cce92

Branch: refs/heads/mr-runner
Commit: ee1cce92d620c78b0243ee0b9c7c0f6ae232b0cb
Parents: c6a3a18
Author: Pei He <pei@apache.org>
Authored: Thu Jul 27 15:05:06 2017 +0800
Committer: Pei He <pei@apache.org>
Committed: Thu Aug 31 14:13:48 2017 +0800

----------------------------------------------------------------------
 runners/map-reduce/pom.xml                      |  2 +-
 .../mapreduce/MapReducePipelineOptions.java     | 51 ++++++++++++++++++++
 runners/pom.xml                                 |  1 +
 3 files changed, 53 insertions(+), 1 deletion(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/ee1cce92/runners/map-reduce/pom.xml
----------------------------------------------------------------------
diff --git a/runners/map-reduce/pom.xml b/runners/map-reduce/pom.xml
index 226c5c0..06e5227 100644
--- a/runners/map-reduce/pom.xml
+++ b/runners/map-reduce/pom.xml
@@ -39,7 +39,7 @@
       <!-- This profile adds execution of ValidatesRunner integration tests
            against a hadoop local cluster. -->
       <id>local-validates-runner-tests</id>
-      <activation><activeByDefault>false</activeByDefault></activation>
+      <activation><activeByDefault>true</activeByDefault></activation>
       <build>
         <plugins>
           <plugin>

http://git-wip-us.apache.org/repos/asf/beam/blob/ee1cce92/runners/map-reduce/src/main/java/org/apache/beam/runners/mapreduce/MapReducePipelineOptions.java
----------------------------------------------------------------------
diff --git a/runners/map-reduce/src/main/java/org/apache/beam/runners/mapreduce/MapReducePipelineOptions.java
b/runners/map-reduce/src/main/java/org/apache/beam/runners/mapreduce/MapReducePipelineOptions.java
index ce8f937..7fe66ba 100644
--- a/runners/map-reduce/src/main/java/org/apache/beam/runners/mapreduce/MapReducePipelineOptions.java
+++ b/runners/map-reduce/src/main/java/org/apache/beam/runners/mapreduce/MapReducePipelineOptions.java
@@ -1,14 +1,65 @@
 package org.apache.beam.runners.mapreduce;
 
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Iterators;
+import java.util.Iterator;
+import java.util.Set;
+import org.apache.beam.sdk.options.Default;
+import org.apache.beam.sdk.options.DefaultValueFactory;
 import org.apache.beam.sdk.options.Description;
 import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
 
 /**
  * {@link PipelineOptions} for {@link MapReduceRunner}.
  */
 public interface MapReducePipelineOptions extends PipelineOptions {
 
+  /** Classes that are used as the boundary in the stack trace to find the callers class
name. */
+  Set<String> PIPELINE_OPTIONS_FACTORY_CLASSES = ImmutableSet.of(
+      PipelineOptionsFactory.class.getName(),
+      PipelineOptionsFactory.Builder.class.getName(),
+      "org.apache.beam.sdk.options.ProxyInvocationHandler");
+
+
   @Description("The jar class of the user Beam program.")
+  @Default.InstanceFactory(JarClassInstanceFactory.class)
   Class<?> getJarClass();
   void setJarClass(Class<?> jarClass);
+
+  class JarClassInstanceFactory implements DefaultValueFactory<Class<?>> {
+    @Override
+    public Class<?> create(PipelineOptions options) {
+      return findCallersClassName(options);
+    }
+
+    /**
+     * Returns the simple name of the calling class using the current threads stack.
+     */
+    private static Class<?> findCallersClassName(PipelineOptions options) {
+      Iterator<StackTraceElement> elements =
+          Iterators.forArray(Thread.currentThread().getStackTrace());
+      // First find the PipelineOptionsFactory/Builder class in the stack trace.
+      while (elements.hasNext()) {
+        StackTraceElement next = elements.next();
+        if (PIPELINE_OPTIONS_FACTORY_CLASSES.contains(next.getClassName())) {
+          break;
+        }
+      }
+      // Then find the first instance after that is not the PipelineOptionsFactory/Builder
class.
+      while (elements.hasNext()) {
+        StackTraceElement next = elements.next();
+        if (!PIPELINE_OPTIONS_FACTORY_CLASSES.contains(next.getClassName())
+            && !next.getClassName().contains("com.sun.proxy.$Proxy")
+            && !next.getClassName().equals(options.getRunner().getName())) {
+          try {
+            return Class.forName(next.getClassName());
+          } catch (ClassNotFoundException e) {
+            break;
+          }
+        }
+      }
+      return null;
+    }
+  }
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/ee1cce92/runners/pom.xml
----------------------------------------------------------------------
diff --git a/runners/pom.xml b/runners/pom.xml
index b00ba9c..4cba41a 100644
--- a/runners/pom.xml
+++ b/runners/pom.xml
@@ -35,6 +35,7 @@
   <modules>
     <module>core-construction-java</module>
     <module>core-java</module>
+    <module>map-reduce</module>
     <module>direct-java</module>
     <module>flink</module>
     <module>google-cloud-dataflow-java</module>


Mime
View raw message