Use random seed for DirectPipelineRunner unless specified

Previously, the DirectPipelineRunner always used the same seed for
pseudorandom operations designed to catch errors in tests that depend
on accidental properties not guaranteed by the Dataflow service.

Now, the seed used is random unless an explicit seed is set
in DirectPipelineOptions.

----Release Notes----

[]
-------------
Created by MOE: http://code.google.com/p/moe-java
MOE_MIGRATED_REVID=99180111

Author: kennknowles

sdk/src/main/java/com/google/cloud/dataflow/sdk/options/DirectPipelineOptions.java

@@ -17,6 +17,9 @@
 package com.google.cloud.dataflow.sdk.options;
 
 import com.google.cloud.dataflow.sdk.runners.DirectPipeline;
+import com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
 
 /**
  * Options that can be used to configure the {@link DirectPipeline}.
@@ -25,4 +28,13 @@ public interface DirectPipelineOptions extends
     ApplicationNameOptions, BigQueryOptions, GcsOptions, GcpOptions,
     PipelineOptions, StreamingOptions {
 
+  /**
+   * The random seed to use for pseudorandom behaviors in the {@link DirectPipelineRunner}.
+   * If not explicitly specified, a random seed will be generated.
+   */
+  @JsonIgnore
+  @Description("The random seed to use for pseudorandom behaviors in the DirectPipelineRunner."
+      + " If not explicitly specified, a random seed will be generated.")
+  Long getDirectPipelineRunnerRandomSeed();
+  void setDirectPipelineRunnerRandomSeed(Long value);
 }

sdk/src/main/java/com/google/cloud/dataflow/sdk/runners/DirectPipelineRunner.java

@@ -79,6 +79,11 @@ public class DirectPipelineRunner
     extends PipelineRunner<DirectPipelineRunner.EvaluationResults> {
   private static final Logger LOG = LoggerFactory.getLogger(DirectPipelineRunner.class);
 
+  /**
+   * A source of random data, which can be seeded if determinism is desired.
+   */
+  private Random rand;
+
   /**
    * A map from PTransform class to the corresponding
    * TransformEvaluator to use to evaluate that transform.
@@ -225,7 +230,7 @@ private <K, InputT, AccumT, OutputT> PCollection<KV<K, OutputT>> applyTestCombin
       PCollection<KV<K, Iterable<InputT>>> input) {
 
     PCollection<KV<K, OutputT>> output = input
-        .apply(ParDo.of(TestCombineDoFn.create(transform, input, testSerializability)));
+        .apply(ParDo.of(TestCombineDoFn.create(transform, input, testSerializability, rand)));
 
     try {
       output.setCoder(transform.getDefaultOutputCoder(input));
@@ -236,30 +241,27 @@ private <K, InputT, AccumT, OutputT> PCollection<KV<K, OutputT>> applyTestCombin
   }
 
   /**
-   * The implementation may split the {@link KeyedCombineFn} into ADD, MERGE
-   * and EXTRACT phases (see {@code com.google.cloud.dataflow.sdk.runners.worker.CombineValuesFn}).
-   * In order to emulate
-   * this for the {@link DirectPipelineRunner} and provide an experience
-   * closer to the service, go through heavy seralizability checks for
-   * the equivalent of the results of the ADD phase, but after the
-   * {@link com.google.cloud.dataflow.sdk.transforms.GroupByKey}
-   * shuffle, and the MERGE phase. Doing these checks
-   * ensure that not only is the accumulator coder serializable, but
-   * the accumulator coder can actually serialize the data in
-   * question.
+   * The implementation may split the {@link KeyedCombineFn} into ADD, MERGE and EXTRACT phases (
+   * see {@code com.google.cloud.dataflow.sdk.runners.worker.CombineValuesFn}). In order to emulate
+   * this for the {@link DirectPipelineRunner} and provide an experience closer to the service, go
+   * through heavy serializability checks for the equivalent of the results of the ADD phase, but
+   * after the {@link com.google.cloud.dataflow.sdk.transforms.GroupByKey} shuffle, and the MERGE
+   * phase. Doing these checks ensure that not only is the accumulator coder serializable, but
+   * the accumulator coder can actually serialize the data in question.
    */
-  // @VisibleForTesting
-  @SuppressWarnings("serial")
   public static class TestCombineDoFn<K, InputT, AccumT, OutputT>
       extends DoFn<KV<K, Iterable<InputT>>, KV<K, OutputT>> {
+    private static final long serialVersionUID = 0L;
     private final KeyedCombineFn<? super K, ? super InputT, AccumT, OutputT> fn;
     private final Coder<AccumT> accumCoder;
     private final boolean testSerializability;
+    private final Random rand;
 
     public static <K, InputT, AccumT, OutputT> TestCombineDoFn<K, InputT, AccumT, OutputT> create(
         Combine.GroupedValues<K, InputT, OutputT> transform,
         PCollection<KV<K, Iterable<InputT>>> input,
-        boolean testSerializability) {
+        boolean testSerializability,
+        Random rand) {
 
       Coder<AccumT> accumCoder;
       try {
@@ -273,16 +275,19 @@ public static <K, InputT, AccumT, OutputT> TestCombineDoFn<K, InputT, AccumT, Ou
       return new TestCombineDoFn(
           transform.getFn(),
           accumCoder,
-          testSerializability);
+          testSerializability,
+          rand);
     }
 
     public TestCombineDoFn(
         KeyedCombineFn<? super K, ? super InputT, AccumT, OutputT> fn,
         Coder<AccumT> accumCoder,
-        boolean testSerializability) {
+        boolean testSerializability,
+        Random rand) {
       this.fn = fn;
       this.accumCoder = accumCoder;
       this.testSerializability = testSerializability;
+      this.rand = rand;
     }
 
     @Override
@@ -291,7 +296,7 @@ public void processElement(ProcessContext c) throws Exception {
       Iterable<InputT> values = c.element().getValue();
       List<AccumT> groupedPostShuffle =
           ensureSerializableByCoder(ListCoder.of(accumCoder),
-              addInputsRandomly(fn, key, values, new Random()),
+              addInputsRandomly(fn, key, values, rand),
               "After addInputs of KeyedCombineFn " + fn.toString());
       AccumT merged =
           ensureSerializableByCoder(accumCoder,
@@ -302,8 +307,11 @@ public void processElement(ProcessContext c) throws Exception {
       c.output(KV.of(key, fn.extractOutput(key, merged)));
     }
 
-    // Create a random list of accumulators from the given list of values
-    // @VisibleForTesting
+    /**
+     * Create a random list of accumulators from the given list of values.
+     *
+     * <p>Visible for testing purposes only.
+     */
     public static <K, AccumT, InputT> List<AccumT> addInputsRandomly(
         KeyedCombineFn<? super K, ? super InputT, AccumT, ?> fn,
         K key,
@@ -352,7 +360,7 @@ public <T> T ensureSerializableByCoder(
   public EvaluationResults run(Pipeline pipeline) {
     LOG.info("Executing pipeline using the DirectPipelineRunner.");
 
-    Evaluator evaluator = new Evaluator();
+    Evaluator evaluator = new Evaluator(rand);
     evaluator.run(pipeline);
 
     // Log all counter values for debugging purposes.
@@ -630,12 +638,15 @@ class Evaluator implements PipelineVisitor, EvaluationContext {
      */
     private final Map<PTransform<?, ?>, String> fullNames = new HashMap<>();
 
-    // Use a random number generator with a fixed seed, so execution
-    // using this evaluator is deterministic.  (If the user-defined
-    // functions, transforms, and coders are deterministic.)
-    Random rand = new Random(0);
+    private Random rand;
 
-    public Evaluator() {}
+    public Evaluator() {
+      this(new Random());
+    }
+
+    public Evaluator(Random rand) {
+      this.rand = rand;
+    }
 
     public void run(Pipeline pipeline) {
       pipeline.traverseTopologically(this);
@@ -953,6 +964,15 @@ private DirectPipelineRunner(DirectPipelineOptions options) {
     this.options = options;
     // (Re-)register standard IO factories. Clobbers any prior credentials.
     IOChannelUtils.registerStandardIOFactories(options);
+    long randomSeed;
+    if (options.getDirectPipelineRunnerRandomSeed() != null) {
+      randomSeed = options.getDirectPipelineRunnerRandomSeed();
+    } else {
+      randomSeed = new Random().nextLong();
+    }
+
+    LOG.info("DirectPipelineRunner using random seed {}.", randomSeed);
+    rand = new Random(randomSeed);
   }
 
   public DirectPipelineOptions getPipelineOptions() {

sdk/src/test/java/com/google/cloud/dataflow/sdk/runners/DirectPipelineRunnerTest.java

@@ -17,9 +17,11 @@
 package com.google.cloud.dataflow.sdk.runners;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
 
 import com.google.cloud.dataflow.sdk.coders.AtomicCoder;
 import com.google.cloud.dataflow.sdk.coders.CoderException;
+import com.google.cloud.dataflow.sdk.options.DirectPipelineOptions;
 import com.google.cloud.dataflow.sdk.options.PipelineOptions;
 import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
 import com.google.cloud.dataflow.sdk.transforms.Create;
@@ -91,4 +93,10 @@ public void processElement(ProcessContext context) {
 
     pipeline.run();
   }
+
+  @Test
+  public void testDirectPipelineOptions() {
+    DirectPipelineOptions options = PipelineOptionsFactory.create().as(DirectPipelineOptions.class);
+    assertNull(options.getDirectPipelineRunnerRandomSeed());
+  }
 }

sdk/src/test/java/com/google/cloud/dataflow/sdk/testing/DataflowAssertTest.java

@@ -18,6 +18,7 @@
 
 import static org.hamcrest.core.StringContains.containsString;
 import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
 import com.google.cloud.dataflow.sdk.Pipeline;
@@ -42,6 +43,7 @@
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.io.Serializable;
+import java.util.regex.Pattern;
 
 /**
  * Test case for {@link DataflowAssert}.
@@ -198,8 +200,12 @@ public void testContainsInAnyOrderFalse() throws Exception {
       try {
         pipeline.run();
       } catch (AssertionError exc) {
-        assertThat(exc.getMessage(),
-            containsString("Expected: iterable over [<4>, <7>, <3>, <2>, <1>] in any order"));
+        // A loose pattern, but should get the job done.
+        Pattern expectedPattern = Pattern.compile(
+            "Expected: iterable over \\[((<4>|<7>|<3>|<2>|<1>)(, )?){5}\\] in any order");
+        assertTrue("Expected error message from DataflowAssert with substring matching "
+            + expectedPattern + " but the message was \"" + exc.getMessage() + "\"",
+            expectedPattern.matcher(exc.getMessage()).find());
         return;
       }
     } else if (pipeline.getRunner() instanceof TestDataflowPipelineRunner) {