beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From da...@apache.org
Subject [01/50] [abbrv] incubator-beam git commit: Remove the DirectPipelineRunner from the Core SDK
Date Mon, 20 Jun 2016 22:15:54 GMT
Repository: incubator-beam
Updated Branches:
  refs/heads/python-sdk 0e5c662b4 -> f166b16b8


http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/99654ca4/sdks/java/core/src/test/java/org/apache/beam/sdk/PipelineTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/PipelineTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/PipelineTest.java
index ea708e5..8abfb05 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/PipelineTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/PipelineTest.java
@@ -28,8 +28,8 @@ import org.apache.beam.sdk.Pipeline.PipelineExecutionException;
 import org.apache.beam.sdk.options.PipelineOptions;
 import org.apache.beam.sdk.options.PipelineOptions.CheckEnabled;
 import org.apache.beam.sdk.options.PipelineOptionsFactory;
-import org.apache.beam.sdk.runners.DirectPipelineRunner;
 import org.apache.beam.sdk.runners.PipelineRunner;
+import org.apache.beam.sdk.testing.CrashingRunner;
 import org.apache.beam.sdk.testing.ExpectedLogs;
 import org.apache.beam.sdk.testing.NeedsRunner;
 import org.apache.beam.sdk.testing.PAssert;
@@ -157,7 +157,7 @@ public class PipelineTest {
   @Test
   public void testToString() {
     PipelineOptions options = PipelineOptionsFactory.as(PipelineOptions.class);
-    options.setRunner(DirectPipelineRunner.class);
+    options.setRunner(CrashingRunner.class);
     Pipeline pipeline = Pipeline.create(options);
     assertEquals("Pipeline#" + pipeline.hashCode(), pipeline.toString());
   }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/99654ca4/sdks/java/core/src/test/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSourceTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSourceTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSourceTest.java
index 774968f..cabfc21 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSourceTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSourceTest.java
@@ -18,14 +18,12 @@
 package org.apache.beam.sdk.io;
 
 import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.includesDisplayDataFrom;
-
 import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 
 import org.apache.beam.sdk.Pipeline;
-import org.apache.beam.sdk.runners.DirectPipelineRunner;
 import org.apache.beam.sdk.runners.dataflow.TestCountingSource;
 import org.apache.beam.sdk.testing.PAssert;
 import org.apache.beam.sdk.testing.RunnableOnService;
@@ -123,10 +121,6 @@ public class BoundedReadFromUnboundedSourceTest implements Serializable{
   private void test(boolean dedup, boolean timeBound) throws Exception {
     Pipeline p = TestPipeline.create();
 
-    if (p.getOptions().getRunner() == DirectPipelineRunner.class) {
-      finalizeTracker = new ArrayList<>();
-      TestCountingSource.setFinalizeTracker(finalizeTracker);
-    }
     TestCountingSource source = new TestCountingSource(Integer.MAX_VALUE).withoutSplitting();
     if (dedup) {
       source = source.withDedup();

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/99654ca4/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/DirectPipelineRegistrarTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/DirectPipelineRegistrarTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/DirectPipelineRegistrarTest.java
deleted file mode 100644
index 92c4835..0000000
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/DirectPipelineRegistrarTest.java
+++ /dev/null
@@ -1,71 +0,0 @@
-/*
- * 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.
- */
-package org.apache.beam.sdk.runners;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
-
-import org.apache.beam.sdk.options.DirectPipelineOptions;
-import org.apache.beam.sdk.options.PipelineOptionsRegistrar;
-
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Lists;
-
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.JUnit4;
-
-import java.util.ServiceLoader;
-
-/** Tests for {@link DirectPipelineRegistrar}. */
-@RunWith(JUnit4.class)
-public class DirectPipelineRegistrarTest {
-  @Test
-  public void testCorrectOptionsAreReturned() {
-    assertEquals(ImmutableList.of(DirectPipelineOptions.class),
-        new DirectPipelineRegistrar.Options().getPipelineOptions());
-  }
-
-  @Test
-  public void testCorrectRunnersAreReturned() {
-    assertEquals(ImmutableList.of(DirectPipelineRunner.class),
-        new DirectPipelineRegistrar.Runner().getPipelineRunners());
-  }
-
-  @Test
-  public void testServiceLoaderForOptions() {
-    for (PipelineOptionsRegistrar registrar :
-        Lists.newArrayList(ServiceLoader.load(PipelineOptionsRegistrar.class).iterator()))
{
-      if (registrar instanceof DirectPipelineRegistrar.Options) {
-        return;
-      }
-    }
-    fail("Expected to find " + DirectPipelineRegistrar.Options.class);
-  }
-
-  @Test
-  public void testServiceLoaderForRunner() {
-    for (PipelineRunnerRegistrar registrar :
-        Lists.newArrayList(ServiceLoader.load(PipelineRunnerRegistrar.class).iterator()))
{
-      if (registrar instanceof DirectPipelineRegistrar.Runner) {
-        return;
-      }
-    }
-    fail("Expected to find " + DirectPipelineRegistrar.Runner.class);
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/99654ca4/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/DirectPipelineRunnerTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/DirectPipelineRunnerTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/DirectPipelineRunnerTest.java
deleted file mode 100644
index edf6996..0000000
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/DirectPipelineRunnerTest.java
+++ /dev/null
@@ -1,222 +0,0 @@
-/*
- * 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.
- */
-package org.apache.beam.sdk.runners;
-
-import static org.hamcrest.Matchers.containsInAnyOrder;
-import static org.hamcrest.Matchers.isA;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertThat;
-
-import org.apache.beam.sdk.Pipeline;
-import org.apache.beam.sdk.coders.AtomicCoder;
-import org.apache.beam.sdk.coders.AvroCoder;
-import org.apache.beam.sdk.coders.Coder;
-import org.apache.beam.sdk.coders.CoderException;
-import org.apache.beam.sdk.io.AvroIO;
-import org.apache.beam.sdk.io.ShardNameTemplate;
-import org.apache.beam.sdk.io.TextIO;
-import org.apache.beam.sdk.options.DirectPipelineOptions;
-import org.apache.beam.sdk.options.PipelineOptions;
-import org.apache.beam.sdk.options.PipelineOptionsFactory;
-import org.apache.beam.sdk.transforms.Create;
-import org.apache.beam.sdk.transforms.DoFn;
-import org.apache.beam.sdk.transforms.ParDo;
-import org.apache.beam.sdk.util.IOChannelUtils;
-
-import com.google.common.collect.Iterables;
-import com.google.common.io.Files;
-
-import org.apache.avro.file.DataFileReader;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.ExpectedException;
-import org.junit.rules.TemporaryFolder;
-import org.junit.runner.RunWith;
-import org.junit.runners.JUnit4;
-
-import java.io.File;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.io.Serializable;
-import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
-import java.util.List;
-
-/** Tests for {@link DirectPipelineRunner}. */
-@RunWith(JUnit4.class)
-public class DirectPipelineRunnerTest implements Serializable {
-  @Rule public TemporaryFolder tmpFolder = new TemporaryFolder();
-  @Rule public ExpectedException expectedException = ExpectedException.none();
-
-  @Test
-  public void testToString() {
-    PipelineOptions options = PipelineOptionsFactory.create();
-    options.setRunner(DirectPipelineRunner.class);
-    DirectPipelineRunner runner = DirectPipelineRunner.fromOptions(options);
-    assertEquals("DirectPipelineRunner#" + runner.hashCode(),
-        runner.toString());
-  }
-
-  /** A {@link Coder} that fails during decoding. */
-  private static class CrashingCoder<T> extends AtomicCoder<T> {
-    @Override
-    public void encode(T value, OutputStream stream, Context context) throws CoderException
{
-      throw new CoderException("Called CrashingCoder.encode");
-    }
-
-    @Override
-    public T decode(
-        InputStream inStream, org.apache.beam.sdk.coders.Coder.Context context)
-            throws CoderException {
-      throw new CoderException("Called CrashingCoder.decode");
-    }
-  }
-
-  /** A {@link DoFn} that outputs {@code 'hello'}. */
-  private static class HelloDoFn extends DoFn<Integer, String> {
-    @Override
-    public void processElement(DoFn<Integer, String>.ProcessContext c) throws Exception
{
-      c.output("hello");
-    }
-  }
-
-  @Test
-  public void testCoderException() {
-    PipelineOptions options = PipelineOptionsFactory.create();
-    options.setRunner(DirectPipelineRunner.class);
-    Pipeline p = Pipeline.create(options);
-
-    p.apply("CreateTestData", Create.of(42))
-        .apply("CrashDuringCoding", ParDo.of(new HelloDoFn()))
-        .setCoder(new CrashingCoder<String>());
-
-    expectedException.expect(RuntimeException.class);
-    expectedException.expectCause(isA(CoderException.class));
-    p.run();
-  }
-
-  @Test
-  public void testDirectPipelineOptions() {
-    DirectPipelineOptions options = PipelineOptionsFactory.create().as(DirectPipelineOptions.class);
-    assertNull(options.getDirectPipelineRunnerRandomSeed());
-  }
-
-  @Test
-  public void testTextIOWriteWithDefaultShardingStrategy() throws Exception {
-    String prefix = IOChannelUtils.resolve(Files.createTempDir().toString(), "output");
-    PipelineOptions options = PipelineOptionsFactory.create();
-    options.setRunner(DirectPipelineRunner.class);
-    Pipeline p = Pipeline.create(options);
-    String[] expectedElements = new String[]{ "a", "b", "c", "d", "e", "f", "g", "h", "i"
};
-    p.apply(Create.of(expectedElements))
-     .apply(TextIO.Write.to(prefix).withSuffix("txt"));
-    p.run();
-
-    String filename =
-        IOChannelUtils.constructName(prefix, ShardNameTemplate.INDEX_OF_MAX, ".txt", 0, 1);
-    List<String> fileContents =
-        Files.readLines(new File(filename), StandardCharsets.UTF_8);
-    // Ensure that each file got at least one record
-    assertFalse(fileContents.isEmpty());
-
-    assertThat(fileContents, containsInAnyOrder(expectedElements));
-  }
-
-  @Test
-  public void testTextIOWriteWithLimitedNumberOfShards() throws Exception {
-    final int numShards = 3;
-    String prefix = IOChannelUtils.resolve(Files.createTempDir().toString(), "shardedOutput");
-    PipelineOptions options = PipelineOptionsFactory.create();
-    options.setRunner(DirectPipelineRunner.class);
-    Pipeline p = Pipeline.create(options);
-    String[] expectedElements = new String[]{ "a", "b", "c", "d", "e", "f", "g", "h", "i"
};
-    p.apply(Create.of(expectedElements))
-     .apply(TextIO.Write.to(prefix).withNumShards(numShards).withSuffix("txt"));
-    p.run();
-
-    List<String> allContents = new ArrayList<>();
-    for (int i = 0; i < numShards; ++i) {
-      String shardFileName =
-          IOChannelUtils.constructName(prefix, ShardNameTemplate.INDEX_OF_MAX, ".txt", i,
3);
-      List<String> shardFileContents =
-          Files.readLines(new File(shardFileName), StandardCharsets.UTF_8);
-
-      // Ensure that each file got at least one record
-      assertFalse(shardFileContents.isEmpty());
-
-      allContents.addAll(shardFileContents);
-    }
-
-    assertThat(allContents, containsInAnyOrder(expectedElements));
-  }
-
-  @Test
-  public void testAvroIOWriteWithDefaultShardingStrategy() throws Exception {
-    String prefix = IOChannelUtils.resolve(Files.createTempDir().toString(), "output");
-    PipelineOptions options = PipelineOptionsFactory.create();
-    options.setRunner(DirectPipelineRunner.class);
-    Pipeline p = Pipeline.create(options);
-    String[] expectedElements = new String[]{ "a", "b", "c", "d", "e", "f", "g", "h", "i"
};
-    p.apply(Create.of(expectedElements))
-     .apply(AvroIO.Write.withSchema(String.class).to(prefix).withSuffix(".avro"));
-    p.run();
-
-    String filename =
-        IOChannelUtils.constructName(prefix, ShardNameTemplate.INDEX_OF_MAX, ".avro", 0,
1);
-    List<String> fileContents = new ArrayList<>();
-    Iterables.addAll(fileContents, DataFileReader.openReader(
-        new File(filename), AvroCoder.of(String.class).createDatumReader()));
-
-    // Ensure that each file got at least one record
-    assertFalse(fileContents.isEmpty());
-
-    assertThat(fileContents, containsInAnyOrder(expectedElements));
-  }
-
-  @Test
-  public void testAvroIOWriteWithLimitedNumberOfShards() throws Exception {
-    final int numShards = 3;
-    String prefix = IOChannelUtils.resolve(Files.createTempDir().toString(), "shardedOutput");
-    PipelineOptions options = PipelineOptionsFactory.create();
-    options.setRunner(DirectPipelineRunner.class);
-    Pipeline p = Pipeline.create(options);
-    String[] expectedElements = new String[]{ "a", "b", "c", "d", "e", "f", "g", "h", "i"
};
-    p.apply(Create.of(expectedElements))
-     .apply(AvroIO.Write.withSchema(String.class).to(prefix)
-                        .withNumShards(numShards).withSuffix(".avro"));
-    p.run();
-
-    List<String> allContents = new ArrayList<>();
-    for (int i = 0; i < numShards; ++i) {
-      String shardFileName =
-          IOChannelUtils.constructName(prefix, ShardNameTemplate.INDEX_OF_MAX, ".avro", i,
3);
-      List<String> shardFileContents = new ArrayList<>();
-      Iterables.addAll(shardFileContents, DataFileReader.openReader(
-          new File(shardFileName), AvroCoder.of(String.class).createDatumReader()));
-
-      // Ensure that each file got at least one record
-      assertFalse(shardFileContents.isEmpty());
-
-      allContents.addAll(shardFileContents);
-    }
-
-    assertThat(allContents, containsInAnyOrder(expectedElements));
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/99654ca4/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/PipelineRunnerTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/PipelineRunnerTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/PipelineRunnerTest.java
index 9313439..5d2e69d 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/PipelineRunnerTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/PipelineRunnerTest.java
@@ -22,6 +22,7 @@ import static org.junit.Assert.assertTrue;
 import org.apache.beam.sdk.options.ApplicationNameOptions;
 import org.apache.beam.sdk.options.DirectPipelineOptions;
 import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.testing.CrashingRunner;
 import org.apache.beam.sdk.util.GcsUtil;
 import org.apache.beam.sdk.util.TestCredential;
 
@@ -53,10 +54,10 @@ public class PipelineRunnerTest {
     options.setAppName("test");
     options.setProject("test");
     options.setGcsUtil(mockGcsUtil);
-    options.setRunner(DirectPipelineRunner.class);
+    options.setRunner(CrashingRunner.class);
     options.setGcpCredential(new TestCredential());
     PipelineRunner<?> runner = PipelineRunner.fromOptions(options);
-    assertTrue(runner instanceof DirectPipelineRunner);
+    assertTrue(runner instanceof CrashingRunner);
   }
 
   @Test
@@ -66,10 +67,10 @@ public class PipelineRunnerTest {
     options.setAppName("test");
     options.setProject("test");
     options.setGcsUtil(mockGcsUtil);
-    options.setRunner(DirectPipelineRunner.class);
+    options.setRunner(CrashingRunner.class);
     options.setGcpCredential(new TestCredential());
     PipelineRunner<?> runner = PipelineRunner.fromOptions(options);
-    assertTrue(runner instanceof DirectPipelineRunner);
+    assertTrue(runner instanceof CrashingRunner);
   }
 
   @Test

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/99654ca4/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CombineTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CombineTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CombineTest.java
index a0b508c..b0ca70b 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CombineTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CombineTest.java
@@ -39,7 +39,6 @@ import org.apache.beam.sdk.coders.KvCoder;
 import org.apache.beam.sdk.coders.SerializableCoder;
 import org.apache.beam.sdk.coders.StandardCoder;
 import org.apache.beam.sdk.coders.StringUtf8Coder;
-import org.apache.beam.sdk.runners.DirectPipelineRunner;
 import org.apache.beam.sdk.testing.NeedsRunner;
 import org.apache.beam.sdk.testing.PAssert;
 import org.apache.beam.sdk.testing.RunnableOnService;
@@ -56,7 +55,6 @@ import org.apache.beam.sdk.transforms.windowing.Sessions;
 import org.apache.beam.sdk.transforms.windowing.SlidingWindows;
 import org.apache.beam.sdk.transforms.windowing.Window;
 import org.apache.beam.sdk.transforms.windowing.Window.ClosingBehavior;
-import org.apache.beam.sdk.util.PerKeyCombineFnRunners;
 import org.apache.beam.sdk.util.PropertyNames;
 import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
 import org.apache.beam.sdk.values.KV;
@@ -87,7 +85,6 @@ import java.util.Arrays;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Objects;
-import java.util.Random;
 import java.util.Set;
 
 /**
@@ -516,24 +513,6 @@ public class CombineTest implements Serializable {
     assertThat(sum.getName(), Matchers.startsWith("Sum"));
   }
 
-  @Test
-  public void testAddInputsRandomly() {
-    TestCounter counter = new TestCounter();
-    Combine.KeyedCombineFn<
-        String, Integer, TestCounter.Counter, Iterable<Long>> fn =
-        counter.asKeyedFn();
-
-    List<TestCounter.Counter> accums = DirectPipelineRunner.TestCombineDoFn.addInputsRandomly(
-        PerKeyCombineFnRunners.create(fn), "bob", Arrays.asList(NUMBERS), new Random(42),
-        processContext);
-
-    assertThat(accums, Matchers.contains(
-        counter.new Counter(3, 2, 0, 0),
-        counter.new Counter(131, 5, 0, 0),
-        counter.new Counter(8, 2, 0, 0),
-        counter.new Counter(1, 1, 0, 0)));
-  }
-
   private static final SerializableFunction<String, Integer> hotKeyFanout =
       new SerializableFunction<String, Integer>() {
         @Override

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/99654ca4/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyTest.java
index 4ce025d..299def7 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyTest.java
@@ -30,9 +30,6 @@ import org.apache.beam.sdk.coders.BigEndianIntegerCoder;
 import org.apache.beam.sdk.coders.KvCoder;
 import org.apache.beam.sdk.coders.MapCoder;
 import org.apache.beam.sdk.coders.StringUtf8Coder;
-import org.apache.beam.sdk.options.DirectPipelineOptions;
-import org.apache.beam.sdk.options.PipelineOptionsFactory;
-import org.apache.beam.sdk.runners.DirectPipelineRunner;
 import org.apache.beam.sdk.testing.NeedsRunner;
 import org.apache.beam.sdk.testing.PAssert;
 import org.apache.beam.sdk.testing.RunnableOnService;
@@ -244,15 +241,9 @@ public class GroupByKeyTest {
                     Duration.standardMinutes(1)))));
   }
 
-  private Pipeline createTestDirectRunner() {
-    DirectPipelineOptions options = PipelineOptionsFactory.as(DirectPipelineOptions.class);
-    options.setRunner(DirectPipelineRunner.class);
-    return Pipeline.create(options);
-  }
-
   @Test
   public void testInvalidWindowsDirect() {
-    Pipeline p = createTestDirectRunner();
+    Pipeline p = TestPipeline.create();
 
     List<KV<String, Integer>> ungroupedPairs = Arrays.asList();
 
@@ -297,7 +288,7 @@ public class GroupByKeyTest {
 
   @Test
   public void testGroupByKeyDirectUnbounded() {
-    Pipeline p = createTestDirectRunner();
+    Pipeline p = TestPipeline.create();
 
     PCollection<KV<String, Integer>> input =
         p.apply(

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/99654ca4/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ViewTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ViewTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ViewTest.java
index 18d39d7..5e6e6a3 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ViewTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ViewTest.java
@@ -33,9 +33,6 @@ import org.apache.beam.sdk.coders.NullableCoder;
 import org.apache.beam.sdk.coders.StringUtf8Coder;
 import org.apache.beam.sdk.coders.VarIntCoder;
 import org.apache.beam.sdk.coders.VoidCoder;
-import org.apache.beam.sdk.options.DirectPipelineOptions;
-import org.apache.beam.sdk.options.PipelineOptionsFactory;
-import org.apache.beam.sdk.runners.DirectPipelineRunner;
 import org.apache.beam.sdk.testing.NeedsRunner;
 import org.apache.beam.sdk.testing.PAssert;
 import org.apache.beam.sdk.testing.RunnableOnService;
@@ -1335,12 +1332,6 @@ public class ViewTest implements Serializable {
     assertEquals("View.AsMultimap", View.<String, Integer>asMultimap().getName());
   }
 
-  private Pipeline createTestDirectRunner() {
-    DirectPipelineOptions options = PipelineOptionsFactory.as(DirectPipelineOptions.class);
-    options.setRunner(DirectPipelineRunner.class);
-    return Pipeline.create(options);
-  }
-
   private void testViewUnbounded(
       Pipeline pipeline,
       PTransform<PCollection<KV<String, Integer>>, ? extends PCollectionView<?>>
view) {
@@ -1378,51 +1369,51 @@ public class ViewTest implements Serializable {
 
   @Test
   public void testViewUnboundedAsSingletonDirect() {
-    testViewUnbounded(createTestDirectRunner(), View.<KV<String, Integer>>asSingleton());
+    testViewUnbounded(TestPipeline.create(), View.<KV<String, Integer>>asSingleton());
   }
 
   @Test
   public void testViewUnboundedAsIterableDirect() {
-    testViewUnbounded(createTestDirectRunner(), View.<KV<String, Integer>>asIterable());
+    testViewUnbounded(TestPipeline.create(), View.<KV<String, Integer>>asIterable());
   }
 
   @Test
   public void testViewUnboundedAsListDirect() {
-    testViewUnbounded(createTestDirectRunner(), View.<KV<String, Integer>>asList());
+    testViewUnbounded(TestPipeline.create(), View.<KV<String, Integer>>asList());
   }
 
   @Test
   public void testViewUnboundedAsMapDirect() {
-    testViewUnbounded(createTestDirectRunner(), View.<String, Integer>asMap());
+    testViewUnbounded(TestPipeline.create(), View.<String, Integer>asMap());
   }
 
   @Test
   public void testViewUnboundedAsMultimapDirect() {
-    testViewUnbounded(createTestDirectRunner(), View.<String, Integer>asMultimap());
+    testViewUnbounded(TestPipeline.create(), View.<String, Integer>asMultimap());
   }
 
   @Test
   public void testViewNonmergingAsSingletonDirect() {
-    testViewNonmerging(createTestDirectRunner(), View.<KV<String, Integer>>asSingleton());
+    testViewNonmerging(TestPipeline.create(), View.<KV<String, Integer>>asSingleton());
   }
 
   @Test
   public void testViewNonmergingAsIterableDirect() {
-    testViewNonmerging(createTestDirectRunner(), View.<KV<String, Integer>>asIterable());
+    testViewNonmerging(TestPipeline.create(), View.<KV<String, Integer>>asIterable());
   }
 
   @Test
   public void testViewNonmergingAsListDirect() {
-    testViewNonmerging(createTestDirectRunner(), View.<KV<String, Integer>>asList());
+    testViewNonmerging(TestPipeline.create(), View.<KV<String, Integer>>asList());
   }
 
   @Test
   public void testViewNonmergingAsMapDirect() {
-    testViewNonmerging(createTestDirectRunner(), View.<String, Integer>asMap());
+    testViewNonmerging(TestPipeline.create(), View.<String, Integer>asMap());
   }
 
   @Test
   public void testViewNonmergingAsMultimapDirect() {
-    testViewNonmerging(createTestDirectRunner(), View.<String, Integer>asMultimap());
+    testViewNonmerging(TestPipeline.create(), View.<String, Integer>asMultimap());
   }
 }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/99654ca4/sdks/java/maven-archetypes/examples/src/main/resources/archetype-resources/src/main/java/common/DataflowExampleUtils.java
----------------------------------------------------------------------
diff --git a/sdks/java/maven-archetypes/examples/src/main/resources/archetype-resources/src/main/java/common/DataflowExampleUtils.java
b/sdks/java/maven-archetypes/examples/src/main/resources/archetype-resources/src/main/java/common/DataflowExampleUtils.java
index 4914d4c..76df4d4 100644
--- a/sdks/java/maven-archetypes/examples/src/main/resources/archetype-resources/src/main/java/common/DataflowExampleUtils.java
+++ b/sdks/java/maven-archetypes/examples/src/main/resources/archetype-resources/src/main/java/common/DataflowExampleUtils.java
@@ -17,6 +17,7 @@
  */
 package ${package}.common;
 
+import org.apache.beam.runners.dataflow.BlockingDataflowPipelineRunner;
 import org.apache.beam.runners.dataflow.DataflowPipelineJob;
 import org.apache.beam.runners.dataflow.DataflowPipelineRunner;
 import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
@@ -39,7 +40,6 @@ import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.PipelineResult;
 import org.apache.beam.sdk.io.TextIO;
 import org.apache.beam.sdk.options.BigQueryOptions;
-import org.apache.beam.sdk.runners.DirectPipelineRunner;
 import org.apache.beam.sdk.transforms.IntraBundleParallelization;
 import org.apache.beam.sdk.util.Transport;
 import com.google.common.collect.Lists;
@@ -250,17 +250,8 @@ public class DataflowExampleUtils {
     }
   }
 
-  /**
-   * Do some runner setup: check that the DirectPipelineRunner is not used in conjunction
with
-   * streaming, and if streaming is specified, use the DataflowPipelineRunner. Return the
streaming
-   * flag value.
-   */
   public void setupRunner() {
-    if (options.isStreaming()) {
-      if (options.getRunner() == DirectPipelineRunner.class) {
-        throw new IllegalArgumentException(
-          "Processing of unbounded input sources is not supported with the DirectPipelineRunner.");
-      }
+    if (options.isStreaming() && options.getRunner().equals(BlockingDataflowPipelineRunner.class))
{
       // In order to cancel the pipelines automatically,
       // {@literal DataflowPipelineRunner} is forced to be used.
       options.setRunner(DataflowPipelineRunner.class);

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/99654ca4/testing/travis/test_wordcount.sh
----------------------------------------------------------------------
diff --git a/testing/travis/test_wordcount.sh b/testing/travis/test_wordcount.sh
index 40e2724..b00b0d6 100755
--- a/testing/travis/test_wordcount.sh
+++ b/testing/travis/test_wordcount.sh
@@ -70,7 +70,7 @@ function run_via_mvn {
   local outfile_prefix="$(get_outfile_prefix "$name")" || exit 2
   local cmd='mvn exec:java -f pom.xml -pl examples/java \
     -Dexec.mainClass=org.apache.beam.examples.WordCount \
-    -Dexec.args="--runner=DirectPipelineRunner --inputFile='"$input"' --output='"$outfile_prefix"'"'
+    -Dexec.args="--runner=InProcessPipelineRunner --inputFile='"$input"' --output='"$outfile_prefix"'"'
   echo "$name: Running $cmd" >&2
   sh -c "$cmd"
   check_result_hash "$name" "$outfile_prefix" "$expected_hash"
@@ -84,7 +84,7 @@ function run_bundled {
   local outfile_prefix="$(get_outfile_prefix "$name")" || exit 2
   local cmd='java -cp '"$JAR_FILE"' \
     org.apache.beam.examples.WordCount \
-    --runner=DirectPipelineRunner \
+    --runner=InProcessPipelineRunner \
     --inputFile='"'$input'"' \
     --output='"$outfile_prefix"
   echo "$name: Running $cmd" >&2


Mime
View raw message