This closes GoogleCloudPlatform#319

bjchambers · May 12, 2016 · d2ad0ed · d2ad0ed
2 parents 2fbc0ea + e406949
commit d2ad0ed
Show file tree

Hide file tree

Showing 5 changed files with 47 additions and 10 deletions.
diff --git a/examples/java/pom.xml b/examples/java/pom.xml
@@ -46,6 +46,17 @@
         <artifactId>maven-compiler-plugin</artifactId>
       </plugin>
 
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-surefire-plugin</artifactId>
+        <configuration>
+          <systemPropertyVariables>
+            <beamTestPipelineOptions>
+            </beamTestPipelineOptions>
+          </systemPropertyVariables>
+        </configuration>
+      </plugin>
+
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-dependency-plugin</artifactId>

diff --git a/runners/direct-java/pom.xml b/runners/direct-java/pom.xml
@@ -81,11 +81,26 @@
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-surefire-plugin</artifactId>
         <executions>
-          <!-- For now, disables integration tests from the SDK as the runner is not ready. -->
           <execution>
             <id>runnable-on-service-tests</id>
+            <phase>integration-test</phase>
+            <goals>
+              <goal>test</goal>
+            </goals>
             <configuration>
-              <skip>true</skip>
+              <groups>org.apache.beam.sdk.testing.RunnableOnService</groups>
+              <parallel>none</parallel>
+              <failIfNoTests>true</failIfNoTests>
+              <dependenciesToScan>
+                <dependency>org.apache.beam:java-sdk-all</dependency>
+              </dependenciesToScan>
+              <systemPropertyVariables>
+                <beamTestPipelineOptions>
+                  [
+                    "--runner=org.apache.beam.runners.direct.InProcessPipelineRunner"
+                  ]
+                </beamTestPipelineOptions>
+              </systemPropertyVariables>
             </configuration>
           </execution>
         </executions>

diff --git a/...chedThreadPoolExecutorServiceFactory.java → ...ixedThreadPoolExecutorServiceFactory.java b/...chedThreadPoolExecutorServiceFactory.java → ...ixedThreadPoolExecutorServiceFactory.java
@@ -24,13 +24,14 @@
 import java.util.concurrent.Executors;
 
 /**
- * A {@link ExecutorServiceFactory} that produces cached thread pools via
- * {@link Executors#newCachedThreadPool()}.
+ * A {@link ExecutorServiceFactory} that produces fixed thread pools via
+ * {@link Executors#newFixedThreadPool(int)}, with the number of threads equal to the available
+ * processors as provided by {@link Runtime#availableProcessors()}.
  */
-class CachedThreadPoolExecutorServiceFactory
+class FixedThreadPoolExecutorServiceFactory
     implements DefaultValueFactory<ExecutorServiceFactory>, ExecutorServiceFactory {
-  private static final CachedThreadPoolExecutorServiceFactory INSTANCE =
-      new CachedThreadPoolExecutorServiceFactory();
+  private static final FixedThreadPoolExecutorServiceFactory INSTANCE =
+      new FixedThreadPoolExecutorServiceFactory();
 
   @Override
   public ExecutorServiceFactory create(PipelineOptions options) {
@@ -39,6 +40,6 @@ public ExecutorServiceFactory create(PipelineOptions options) {
 
   @Override
   public ExecutorService create() {
-    return Executors.newCachedThreadPool();
+    return Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
   }
 }
diff --git a/...rs/direct-java/src/main/java/org/apache/beam/runners/direct/InProcessPipelineOptions.java b/...rs/direct-java/src/main/java/org/apache/beam/runners/direct/InProcessPipelineOptions.java
@@ -43,13 +43,13 @@ public interface InProcessPipelineOptions extends PipelineOptions, ApplicationNa
    * it cannot enter a state in which it will not schedule additional pending work unless currently
    * scheduled work completes, as this may cause the {@link Pipeline} to cease processing.
    *
-   * <p>Defaults to a {@link CachedThreadPoolExecutorServiceFactory}, which produces instances of
+   * <p>Defaults to a {@link FixedThreadPoolExecutorServiceFactory}, which produces instances of
    * {@link Executors#newCachedThreadPool()}.
    */
   @JsonIgnore
   @Required
   @Hidden
-  @Default.InstanceFactory(CachedThreadPoolExecutorServiceFactory.class)
+  @Default.InstanceFactory(FixedThreadPoolExecutorServiceFactory.class)
   ExecutorServiceFactory getExecutorServiceFactory();
 
   void setExecutorServiceFactory(ExecutorServiceFactory executorService);

diff --git a/sdks/java/core/pom.xml b/sdks/java/core/pom.xml
@@ -124,6 +124,16 @@
         <artifactId>maven-compiler-plugin</artifactId>
       </plugin>
 
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-surefire-plugin</artifactId>
+        <configuration>
+          <excludedGroups>
+            org.apache.beam.sdk.testing.RunnableOnService
+          </excludedGroups>
+        </configuration>
+      </plugin>
+
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-dependency-plugin</artifactId>