Merge pull request GoogleCloudPlatform#438 from tgroh/backport_960

Add Flatten with Duplicate Inputs Test

sdk/src/test/java/com/google/cloud/dataflow/sdk/transforms/FlattenTest.java

@@ -40,7 +40,7 @@
 import com.google.cloud.dataflow.sdk.values.PCollectionList;
 import com.google.cloud.dataflow.sdk.values.PCollectionView;
 import com.google.common.collect.ImmutableSet;
-
+import com.google.common.collect.Iterables;
 import org.joda.time.Duration;
 import org.junit.Assert;
 import org.junit.Rule;
@@ -49,7 +49,6 @@
 import org.junit.rules.ExpectedException;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
-
 import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -115,6 +114,26 @@ public void testFlattenPCollectionListEmpty() {
     p.run();
   }
 
+  @Test
+  @Category(RunnableOnService.class)
+  public void testFlattenInputMultipleCopies() {
+    Pipeline p = TestPipeline.create();
+
+    PCollection<String> lines = p.apply("mkLines", Create.of(LINES));
+    PCollection<String> lines2 = p.apply("mkOtherLines", Create.of(LINES2));
+
+    PCollection<String> flattened = PCollectionList.of(lines)
+        .and(lines2)
+        .and(lines)
+        .and(lines)
+        .apply(Flatten.<String>pCollections());
+
+    DataflowAssert.that(flattened)
+        .containsInAnyOrder(Iterables.concat(LINES, LINES, LINES, LINES2));
+
+    p.run();
+  }
+
   @Test
   @Category(RunnableOnService.class)
   public void testEmptyFlattenAsSideInput() {