Switch to using a FixedThreadPool in the InProcessPipelineRunner

sdk/src/main/java/com/google/cloud/dataflow/sdk/runners/inprocess/CachedThreadPoolExecutorServiceFactory.java → sdk/src/main/java/com/google/cloud/dataflow/sdk/runners/inprocess/FixedThreadPoolExecutorServiceFactory.java

@@ -22,13 +22,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) {
@@ -37,6 +38,6 @@ public ExecutorServiceFactory create(PipelineOptions options) {
 
   @Override
   public ExecutorService create() {
-    return Executors.newCachedThreadPool();
+    return Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
   }
 }

sdk/src/main/java/com/google/cloud/dataflow/sdk/runners/inprocess/InProcessPipelineOptions.java

@@ -41,13 +41,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);