beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From tg...@apache.org
Subject [2/3] incubator-beam git commit: Move Towards removing WindowedValue from SDK
Date Fri, 02 Dec 2016 21:17:19 GMT
Move Towards removing WindowedValue from SDK

- Introduces ValueInSingleWindow for purposes of PAssert
- Uses ValueInSingleWindow inside DoFnTester
- Moves WindowMatchers{,Test} to runners-core

After this commit, WindowedValue does not appear in any SDK APIs
used by Pipeline authors.


Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/d9891234
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/d9891234
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/d9891234

Branch: refs/heads/master
Commit: d989123424a54699ecb47ba6c0a4e437316cabce
Parents: 0fb5610
Author: Eugene Kirpichov <kirpichov@google.com>
Authored: Mon Oct 31 15:46:25 2016 -0700
Committer: Thomas Groh <tgroh@google.com>
Committed: Fri Dec 2 13:16:04 2016 -0800

----------------------------------------------------------------------
 .../beam/runners/core/ReduceFnRunnerTest.java   |   5 +-
 .../beam/runners/core/SplittableParDoTest.java  |  38 ++--
 .../beam/runners/core/WindowMatchers.java       | 204 +++++++++++++++++++
 .../beam/runners/core/WindowMatchersTest.java   |  82 ++++++++
 .../direct/WindowEvaluatorFactoryTest.java      |   4 +-
 .../apache/beam/sdk/testing/GatherAllPanes.java |  88 ++++++++
 .../org/apache/beam/sdk/testing/PAssert.java    |  77 +++----
 .../apache/beam/sdk/testing/PaneExtractors.java |  55 +++--
 .../beam/sdk/testing/ValueInSingleWindow.java   | 134 ++++++++++++
 .../apache/beam/sdk/transforms/DoFnTester.java  |  58 +++---
 .../apache/beam/sdk/util/GatherAllPanes.java    |  86 --------
 .../apache/beam/sdk/util/IdentityWindowFn.java  |   2 +-
 .../org/apache/beam/sdk/WindowMatchers.java     | 204 -------------------
 .../org/apache/beam/sdk/WindowMatchersTest.java |  82 --------
 .../beam/sdk/testing/GatherAllPanesTest.java    | 140 +++++++++++++
 .../beam/sdk/testing/PaneExtractorsTest.java    | 133 ++++++------
 .../testing/ValueInSingleWindowCoderTest.java   |  51 +++++
 .../beam/sdk/util/GatherAllPanesTest.java       | 143 -------------
 18 files changed, 893 insertions(+), 693 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d9891234/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnRunnerTest.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnRunnerTest.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnRunnerTest.java
index 20eb08b..ba57567 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnRunnerTest.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnRunnerTest.java
@@ -18,8 +18,8 @@
 package org.apache.beam.runners.core;
 
 import static com.google.common.base.Preconditions.checkArgument;
-import static org.apache.beam.sdk.WindowMatchers.isSingleWindowedValue;
-import static org.apache.beam.sdk.WindowMatchers.isWindowedValue;
+import static org.apache.beam.runners.core.WindowMatchers.isSingleWindowedValue;
+import static org.apache.beam.runners.core.WindowMatchers.isWindowedValue;
 import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.emptyIterable;
@@ -39,7 +39,6 @@ import com.google.common.collect.Iterables;
 import java.util.Iterator;
 import java.util.List;
 import org.apache.beam.runners.core.triggers.TriggerStateMachine;
-import org.apache.beam.sdk.WindowMatchers;
 import org.apache.beam.sdk.coders.VarIntCoder;
 import org.apache.beam.sdk.options.PipelineOptions;
 import org.apache.beam.sdk.options.PipelineOptionsFactory;

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d9891234/runners/core-java/src/test/java/org/apache/beam/runners/core/SplittableParDoTest.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/SplittableParDoTest.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/SplittableParDoTest.java
index 990d892..b13d839 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/SplittableParDoTest.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/SplittableParDoTest.java
@@ -36,6 +36,7 @@ import org.apache.beam.sdk.coders.BigEndianIntegerCoder;
 import org.apache.beam.sdk.coders.Coder;
 import org.apache.beam.sdk.coders.SerializableCoder;
 import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.ValueInSingleWindow;
 import org.apache.beam.sdk.transforms.Create;
 import org.apache.beam.sdk.transforms.DoFn;
 import org.apache.beam.sdk.transforms.DoFnTester;
@@ -142,14 +143,15 @@ public class SplittableParDoTest {
         PCollection.IsBounded.BOUNDED,
         makeBoundedCollection(pipeline)
             .apply("bounded to bounded", new SplittableParDo<>(makeParDo(boundedFn)))
-            .get(MAIN_OUTPUT_TAG).isBounded());
+            .get(MAIN_OUTPUT_TAG)
+            .isBounded());
     assertEquals(
         "Applying a bounded SDF to an unbounded collection produces an unbounded collection",
         PCollection.IsBounded.UNBOUNDED,
         makeUnboundedCollection(pipeline)
-            .apply(
-                "bounded to unbounded", new SplittableParDo<>(makeParDo(boundedFn)))
-            .get(MAIN_OUTPUT_TAG).isBounded());
+            .apply("bounded to unbounded", new SplittableParDo<>(makeParDo(boundedFn)))
+            .get(MAIN_OUTPUT_TAG)
+            .isBounded());
   }
 
   @Test
@@ -160,18 +162,16 @@ public class SplittableParDoTest {
         "Applying an unbounded SDF to a bounded collection produces a bounded collection",
         PCollection.IsBounded.UNBOUNDED,
         makeBoundedCollection(pipeline)
-            .apply(
-                "unbounded to bounded",
-                new SplittableParDo<>(makeParDo(unboundedFn)))
-            .get(MAIN_OUTPUT_TAG).isBounded());
+            .apply("unbounded to bounded", new SplittableParDo<>(makeParDo(unboundedFn)))
+            .get(MAIN_OUTPUT_TAG)
+            .isBounded());
     assertEquals(
         "Applying an unbounded SDF to an unbounded collection produces an unbounded collection",
         PCollection.IsBounded.UNBOUNDED,
         makeUnboundedCollection(pipeline)
-            .apply(
-                "unbounded to unbounded",
-                new SplittableParDo<>(makeParDo(unboundedFn)))
-            .get(MAIN_OUTPUT_TAG).isBounded());
+            .apply("unbounded to unbounded", new SplittableParDo<>(makeParDo(unboundedFn)))
+            .get(MAIN_OUTPUT_TAG)
+            .isBounded());
   }
 
   // ------------------------------- Tests for ProcessFn ---------------------------------
@@ -224,9 +224,11 @@ public class SplittableParDoTest {
                 Instant timestamp,
                 Collection<? extends BoundedWindow> windows,
                 PaneInfo pane) {
-              tester
-                  .getMutableOutput(tester.getMainOutputTag())
-                  .add(WindowedValue.of(output, timestamp, windows, pane));
+              for (BoundedWindow window : windows) {
+                tester
+                    .getMutableOutput(tester.getMainOutputTag())
+                    .add(ValueInSingleWindow.of(output, timestamp, window, pane));
+              }
             }
 
             @Override
@@ -236,7 +238,11 @@ public class SplittableParDoTest {
                 Instant timestamp,
                 Collection<? extends BoundedWindow> windows,
                 PaneInfo pane) {
-              tester.getMutableOutput(tag).add(WindowedValue.of(output, timestamp, windows, pane));
+              for (BoundedWindow window : windows) {
+                tester
+                    .getMutableOutput(tag)
+                    .add(ValueInSingleWindow.of(output, timestamp, window, pane));
+              }
             }
           });
       // Do not clone since ProcessFn references non-serializable DoFnTester itself

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d9891234/runners/core-java/src/test/java/org/apache/beam/runners/core/WindowMatchers.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/WindowMatchers.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/WindowMatchers.java
new file mode 100644
index 0000000..6c3a7e2
--- /dev/null
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/WindowMatchers.java
@@ -0,0 +1,204 @@
+/*
+ * 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.runners.core;
+
+import com.google.common.collect.Lists;
+import java.util.Collection;
+import java.util.Objects;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
+import org.apache.beam.sdk.transforms.windowing.PaneInfo;
+import org.apache.beam.sdk.util.WindowedValue;
+import org.hamcrest.Description;
+import org.hamcrest.Matcher;
+import org.hamcrest.Matchers;
+import org.hamcrest.TypeSafeMatcher;
+import org.joda.time.Instant;
+
+/**
+ * Matchers that are useful for working with Windowing, Timestamps, etc.
+ */
+public class WindowMatchers {
+
+  public static <T> Matcher<WindowedValue<? extends T>> isWindowedValue(
+      T value,
+      Instant timestamp,
+      Collection<? extends BoundedWindow> windows,
+      PaneInfo paneInfo) {
+
+    Collection<Matcher<? super BoundedWindow>> windowMatchers =
+        Lists.newArrayListWithCapacity(windows.size());
+    for (BoundedWindow window : windows) {
+      windowMatchers.add(Matchers.equalTo(window));
+    }
+
+    return isWindowedValue(
+        Matchers.equalTo(value),
+        Matchers.equalTo(timestamp),
+        Matchers.containsInAnyOrder(windowMatchers),
+        Matchers.equalTo(paneInfo));
+  }
+
+  public static <T> Matcher<WindowedValue<? extends T>> isWindowedValue(
+      Matcher<? super T> valueMatcher,
+      Matcher<? super Instant> timestampMatcher,
+      Matcher<? super Collection<? extends BoundedWindow>> windowsMatcher,
+      Matcher<? super PaneInfo> paneInfoMatcher) {
+    return new WindowedValueMatcher<>(
+        valueMatcher, timestampMatcher, windowsMatcher, paneInfoMatcher);
+  }
+
+  public static <T> Matcher<WindowedValue<? extends T>> isWindowedValue(
+      Matcher<? super T> valueMatcher,
+      Matcher<? super Instant> timestampMatcher,
+      Matcher<? super Collection<? extends BoundedWindow>> windowsMatcher) {
+    return new WindowedValueMatcher<>(
+        valueMatcher, timestampMatcher, windowsMatcher, Matchers.anything());
+  }
+
+  public static <T> Matcher<WindowedValue<? extends T>> isWindowedValue(
+      Matcher<? super T> valueMatcher, Matcher<? super Instant> timestampMatcher) {
+    return new WindowedValueMatcher<>(
+        valueMatcher, timestampMatcher, Matchers.anything(), Matchers.anything());
+  }
+
+  public static <T> Matcher<WindowedValue<? extends T>> isWindowedValue(
+      Matcher<? super T> valueMatcher) {
+    return new WindowedValueMatcher<>(
+        valueMatcher, Matchers.anything(), Matchers.anything(), Matchers.anything());
+  }
+
+  public static <T> Matcher<WindowedValue<? extends T>> isSingleWindowedValue(
+      T value, long timestamp, long windowStart, long windowEnd) {
+    return WindowMatchers.<T>isSingleWindowedValue(
+        Matchers.equalTo(value), timestamp, windowStart, windowEnd);
+  }
+
+  public static <T> Matcher<WindowedValue<? extends T>> isSingleWindowedValue(
+      T value, Instant timestamp, BoundedWindow window, PaneInfo paneInfo) {
+    return WindowMatchers.<T>isSingleWindowedValue(
+        Matchers.equalTo(value),
+        Matchers.equalTo(timestamp),
+        Matchers.equalTo(window),
+        Matchers.equalTo(paneInfo));
+  }
+
+  public static <T> Matcher<WindowedValue<? extends T>> isSingleWindowedValue(
+      T value, Instant timestamp, BoundedWindow window) {
+    return WindowMatchers.<T>isSingleWindowedValue(
+        Matchers.equalTo(value), Matchers.equalTo(timestamp), Matchers.equalTo(window));
+  }
+
+  public static <T> Matcher<WindowedValue<? extends T>> isSingleWindowedValue(
+      Matcher<T> valueMatcher, long timestamp, long windowStart, long windowEnd) {
+    IntervalWindow intervalWindow =
+        new IntervalWindow(new Instant(windowStart), new Instant(windowEnd));
+    return WindowMatchers.<T>isSingleWindowedValue(
+        valueMatcher,
+        Matchers.describedAs("%0", Matchers.equalTo(new Instant(timestamp)), timestamp),
+        Matchers.<BoundedWindow>equalTo(intervalWindow),
+        Matchers.anything());
+  }
+
+  public static <T> Matcher<WindowedValue<? extends T>> isSingleWindowedValue(
+      Matcher<? super T> valueMatcher,
+      Matcher<? super Instant> timestampMatcher,
+      Matcher<? super BoundedWindow> windowMatcher) {
+    return new WindowedValueMatcher<T>(
+        valueMatcher, timestampMatcher, Matchers.contains(windowMatcher), Matchers.anything());
+  }
+
+  public static <T> Matcher<WindowedValue<? extends T>> isSingleWindowedValue(
+      Matcher<? super T> valueMatcher,
+      Matcher<? super Instant> timestampMatcher,
+      Matcher<? super BoundedWindow> windowMatcher,
+      Matcher<? super PaneInfo> paneInfoMatcher) {
+    return new WindowedValueMatcher<T>(
+        valueMatcher, timestampMatcher, Matchers.contains(windowMatcher), paneInfoMatcher);
+  }
+
+  public static Matcher<IntervalWindow> intervalWindow(long start, long end) {
+    return Matchers.equalTo(new IntervalWindow(new Instant(start), new Instant(end)));
+  }
+
+  public static <T> Matcher<WindowedValue<? extends T>> valueWithPaneInfo(final PaneInfo paneInfo) {
+    return new TypeSafeMatcher<WindowedValue<? extends T>>() {
+      @Override
+      public void describeTo(Description description) {
+        description
+            .appendText("WindowedValue(paneInfo = ").appendValue(paneInfo).appendText(")");
+      }
+
+      @Override
+      protected boolean matchesSafely(WindowedValue<? extends T> item) {
+        return Objects.equals(item.getPane(), paneInfo);
+      }
+
+      @Override
+      protected void describeMismatchSafely(
+          WindowedValue<? extends T> item, Description mismatchDescription) {
+        mismatchDescription.appendValue(item.getPane());
+      }
+    };
+  }
+
+  @SuppressWarnings({"unchecked", "rawtypes"})
+  @SafeVarargs
+  public static final <W extends BoundedWindow> Matcher<Iterable<W>> ofWindows(
+      Matcher<W>... windows) {
+    return (Matcher) Matchers.<W>containsInAnyOrder(windows);
+  }
+
+  private WindowMatchers() {}
+
+  private static class WindowedValueMatcher<T> extends TypeSafeMatcher<WindowedValue<? extends T>> {
+
+    private Matcher<? super T> valueMatcher;
+    private Matcher<? super Instant> timestampMatcher;
+    private Matcher<? super Collection<? extends BoundedWindow>> windowsMatcher;
+    private Matcher<? super PaneInfo> paneInfoMatcher;
+
+    private WindowedValueMatcher(
+        Matcher<? super T> valueMatcher,
+        Matcher<? super Instant> timestampMatcher,
+        Matcher<? super Collection<? extends BoundedWindow>> windowsMatcher,
+        Matcher<? super PaneInfo> paneInfoMatcher) {
+      this.valueMatcher = valueMatcher;
+      this.timestampMatcher = timestampMatcher;
+      this.windowsMatcher = windowsMatcher;
+      this.paneInfoMatcher = paneInfoMatcher;
+    }
+
+    @Override
+    public void describeTo(Description description) {
+      description
+          .appendText("a WindowedValue(").appendValue(valueMatcher)
+          .appendText(", ").appendValue(timestampMatcher)
+          .appendText(", ").appendValue(windowsMatcher)
+          .appendText(", ").appendValue(paneInfoMatcher)
+          .appendText(")");
+    }
+
+    @Override
+    protected boolean matchesSafely(WindowedValue<? extends T> windowedValue) {
+      return valueMatcher.matches(windowedValue.getValue())
+          && timestampMatcher.matches(windowedValue.getTimestamp())
+          && windowsMatcher.matches(windowedValue.getWindows());
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d9891234/runners/core-java/src/test/java/org/apache/beam/runners/core/WindowMatchersTest.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/WindowMatchersTest.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/WindowMatchersTest.java
new file mode 100644
index 0000000..6f4741a
--- /dev/null
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/WindowMatchersTest.java
@@ -0,0 +1,82 @@
+/*
+ * 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.runners.core;
+
+import static org.junit.Assert.assertThat;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
+import org.apache.beam.sdk.transforms.windowing.PaneInfo;
+import org.apache.beam.sdk.util.WindowedValue;
+import org.joda.time.Instant;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/**
+ * Tests for {@link WindowMatchers}.
+ */
+@RunWith(JUnit4.class)
+public class WindowMatchersTest {
+
+  @Test
+  public void testIsWindowedValueExact() {
+    long timestamp = 100;
+    long windowStart = 0;
+    long windowEnd = 200;
+
+    assertThat(
+        WindowedValue.of(
+            "hello",
+            new Instant(timestamp),
+            new IntervalWindow(new Instant(windowStart), new Instant(windowEnd)),
+            PaneInfo.NO_FIRING),
+        WindowMatchers.isWindowedValue(
+            "hello",
+            new Instant(timestamp),
+            ImmutableList.of(new IntervalWindow(new Instant(windowStart), new Instant(windowEnd))),
+            PaneInfo.NO_FIRING));
+  }
+
+  @Test
+  public void testIsWindowedValueReorderedWindows() {
+    long timestamp = 100;
+    long windowStart = 0;
+    long windowEnd = 200;
+    long windowStart2 = 50;
+    long windowEnd2 = 150;
+
+    assertThat(
+        WindowedValue.of(
+            "hello",
+            new Instant(timestamp),
+            ImmutableList.of(
+                new IntervalWindow(new Instant(windowStart), new Instant(windowEnd)),
+                new IntervalWindow(new Instant(windowStart2), new Instant(windowEnd2))),
+            PaneInfo.NO_FIRING),
+        WindowMatchers.isWindowedValue(
+            "hello",
+            new Instant(timestamp),
+            ImmutableList.of(
+                new IntervalWindow(new Instant(windowStart), new Instant(windowEnd)),
+                new IntervalWindow(new Instant(windowStart2), new Instant(windowEnd2))),
+            PaneInfo.NO_FIRING));
+  }
+}
+
+

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d9891234/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WindowEvaluatorFactoryTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WindowEvaluatorFactoryTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WindowEvaluatorFactoryTest.java
index e2f987c..66c28ce 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WindowEvaluatorFactoryTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WindowEvaluatorFactoryTest.java
@@ -17,8 +17,8 @@
  */
 package org.apache.beam.runners.direct;
 
-import static org.apache.beam.sdk.WindowMatchers.isSingleWindowedValue;
-import static org.apache.beam.sdk.WindowMatchers.isWindowedValue;
+import static org.apache.beam.runners.core.WindowMatchers.isSingleWindowedValue;
+import static org.apache.beam.runners.core.WindowMatchers.isWindowedValue;
 import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.junit.Assert.assertThat;
 import static org.mockito.Mockito.when;

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d9891234/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/GatherAllPanes.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/GatherAllPanes.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/GatherAllPanes.java
new file mode 100644
index 0000000..2b311b7
--- /dev/null
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/GatherAllPanes.java
@@ -0,0 +1,88 @@
+/*
+ * 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.testing;
+
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Values;
+import org.apache.beam.sdk.transforms.WithKeys;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.Never;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.transforms.windowing.WindowFn;
+import org.apache.beam.sdk.util.IdentityWindowFn;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TypeDescriptor;
+
+/**
+ * Gathers all panes of each window into exactly one output.
+ *
+ * <p>Note that this will delay the output of a window until the garbage collection time (when the
+ * watermark passes the end of the window plus allowed lateness) even if the upstream triggers
+ * closed the window earlier.
+ */
+class GatherAllPanes<T>
+    extends PTransform<PCollection<T>, PCollection<Iterable<ValueInSingleWindow<T>>>> {
+  /**
+   * Gathers all panes of each window into a single output element.
+   *
+   * <p>This will gather all output panes into a single element, which causes them to be colocated
+   * on a single worker. As a result, this is only suitable for {@link PCollection PCollections}
+   * where all of the output elements for each pane fit in memory, such as in tests.
+   */
+  public static <T> GatherAllPanes<T> globally() {
+    return new GatherAllPanes<>();
+  }
+
+  private GatherAllPanes() {}
+
+  @Override
+  public PCollection<Iterable<ValueInSingleWindow<T>>> apply(PCollection<T> input) {
+    WindowFn<?, ?> originalWindowFn = input.getWindowingStrategy().getWindowFn();
+
+    return input
+        .apply(ParDo.of(new ReifyTimestampsAndWindowsFn<T>()))
+        .setCoder(
+            ValueInSingleWindow.Coder.of(
+                input.getCoder(), input.getWindowingStrategy().getWindowFn().windowCoder()))
+        .apply(
+            WithKeys.<Integer, ValueInSingleWindow<T>>of(0)
+                .withKeyType(new TypeDescriptor<Integer>() {}))
+        .apply(
+            Window.into(
+                    new IdentityWindowFn<KV<Integer, ValueInSingleWindow<T>>>(
+                        originalWindowFn.windowCoder()))
+                .triggering(Never.ever())
+                .withAllowedLateness(input.getWindowingStrategy().getAllowedLateness())
+                .discardingFiredPanes())
+        // all values have the same key so they all appear as a single output element
+        .apply(GroupByKey.<Integer, ValueInSingleWindow<T>>create())
+        .apply(Values.<Iterable<ValueInSingleWindow<T>>>create())
+        .setWindowingStrategyInternal(input.getWindowingStrategy());
+  }
+
+  private static class ReifyTimestampsAndWindowsFn<T> extends DoFn<T, ValueInSingleWindow<T>> {
+    @DoFn.ProcessElement
+    public void processElement(ProcessContext c, BoundedWindow window) {
+      c.output(ValueInSingleWindow.of(c.element(), c.timestamp(), window, c.pane()));
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d9891234/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/PAssert.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/PAssert.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/PAssert.java
index b3a14aa..7dc78d8 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/PAssert.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/PAssert.java
@@ -63,8 +63,6 @@ import org.apache.beam.sdk.transforms.windowing.Trigger;
 import org.apache.beam.sdk.transforms.windowing.Window;
 import org.apache.beam.sdk.transforms.windowing.Window.ClosingBehavior;
 import org.apache.beam.sdk.util.CoderUtils;
-import org.apache.beam.sdk.util.GatherAllPanes;
-import org.apache.beam.sdk.util.WindowedValue;
 import org.apache.beam.sdk.util.WindowingStrategy;
 import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.PBegin;
@@ -349,7 +347,7 @@ public class PAssert {
   private static class PCollectionContentsAssert<T> implements IterableAssert<T> {
     private final PCollection<T> actual;
     private final AssertionWindows rewindowingStrategy;
-    private final SimpleFunction<Iterable<WindowedValue<T>>, Iterable<T>> paneExtractor;
+    private final SimpleFunction<Iterable<ValueInSingleWindow<T>>, Iterable<T>> paneExtractor;
 
     public PCollectionContentsAssert(PCollection<T> actual) {
       this(actual, IntoGlobalWindow.<T>of(), PaneExtractors.<T>allPanes());
@@ -358,7 +356,7 @@ public class PAssert {
     public PCollectionContentsAssert(
         PCollection<T> actual,
         AssertionWindows rewindowingStrategy,
-        SimpleFunction<Iterable<WindowedValue<T>>, Iterable<T>> paneExtractor) {
+        SimpleFunction<Iterable<ValueInSingleWindow<T>>, Iterable<T>> paneExtractor) {
       this.actual = actual;
       this.rewindowingStrategy = rewindowingStrategy;
       this.paneExtractor = paneExtractor;
@@ -391,7 +389,7 @@ public class PAssert {
 
     private PCollectionContentsAssert<T> withPane(
         BoundedWindow window,
-        SimpleFunction<Iterable<WindowedValue<T>>, Iterable<T>> paneExtractor) {
+        SimpleFunction<Iterable<ValueInSingleWindow<T>>, Iterable<T>> paneExtractor) {
       @SuppressWarnings({"unchecked", "rawtypes"})
       Coder<BoundedWindow> windowCoder =
           (Coder) actual.getWindowingStrategy().getWindowFn().windowCoder();
@@ -523,7 +521,7 @@ public class PAssert {
     private final PCollection<Iterable<T>> actual;
     private final Coder<T> elementCoder;
     private final AssertionWindows rewindowingStrategy;
-    private final SimpleFunction<Iterable<WindowedValue<Iterable<T>>>, Iterable<Iterable<T>>>
+    private final SimpleFunction<Iterable<ValueInSingleWindow<Iterable<T>>>, Iterable<Iterable<T>>>
         paneExtractor;
 
     public PCollectionSingletonIterableAssert(PCollection<Iterable<T>> actual) {
@@ -533,7 +531,8 @@ public class PAssert {
     public PCollectionSingletonIterableAssert(
         PCollection<Iterable<T>> actual,
         AssertionWindows rewindowingStrategy,
-        SimpleFunction<Iterable<WindowedValue<Iterable<T>>>, Iterable<Iterable<T>>> paneExtractor) {
+        SimpleFunction<Iterable<ValueInSingleWindow<Iterable<T>>>, Iterable<Iterable<T>>>
+            paneExtractor) {
       this.actual = actual;
 
       @SuppressWarnings("unchecked")
@@ -571,7 +570,8 @@ public class PAssert {
 
     private PCollectionSingletonIterableAssert<T> withPanes(
         BoundedWindow window,
-        SimpleFunction<Iterable<WindowedValue<Iterable<T>>>, Iterable<Iterable<T>>> paneExtractor) {
+        SimpleFunction<Iterable<ValueInSingleWindow<Iterable<T>>>, Iterable<Iterable<T>>>
+            paneExtractor) {
       @SuppressWarnings({"unchecked", "rawtypes"})
       Coder<BoundedWindow> windowCoder =
           (Coder) actual.getWindowingStrategy().getWindowFn().windowCoder();
@@ -620,7 +620,8 @@ public class PAssert {
     private final PCollection<ElemT> actual;
     private final PTransform<PCollection<ElemT>, PCollectionView<ViewT>> view;
     private final AssertionWindows rewindowActuals;
-    private final SimpleFunction<Iterable<WindowedValue<ElemT>>, Iterable<ElemT>> paneExtractor;
+    private final SimpleFunction<Iterable<ValueInSingleWindow<ElemT>>, Iterable<ElemT>>
+        paneExtractor;
     private final Coder<ViewT> coder;
 
     protected PCollectionViewAssert(
@@ -634,7 +635,7 @@ public class PAssert {
         PCollection<ElemT> actual,
         PTransform<PCollection<ElemT>, PCollectionView<ViewT>> view,
         AssertionWindows rewindowActuals,
-        SimpleFunction<Iterable<WindowedValue<ElemT>>, Iterable<ElemT>> paneExtractor,
+        SimpleFunction<Iterable<ValueInSingleWindow<ElemT>>, Iterable<ElemT>> paneExtractor,
         Coder<ViewT> coder) {
       this.actual = actual;
       this.view = view;
@@ -660,7 +661,7 @@ public class PAssert {
 
     private PCollectionViewAssert<ElemT, ViewT> inPane(
         BoundedWindow window,
-        SimpleFunction<Iterable<WindowedValue<ElemT>>, Iterable<ElemT>> paneExtractor) {
+        SimpleFunction<Iterable<ValueInSingleWindow<ElemT>>, Iterable<ElemT>> paneExtractor) {
       return new PCollectionViewAssert<>(
           actual,
           view,
@@ -738,13 +739,14 @@ public class PAssert {
 
     private final transient PCollection<T> actual;
     private final transient AssertionWindows rewindowActuals;
-    private final transient SimpleFunction<Iterable<WindowedValue<T>>, Iterable<T>> extractPane;
+    private final transient SimpleFunction<Iterable<ValueInSingleWindow<T>>, Iterable<T>>
+        extractPane;
     private final transient PTransform<PCollection<T>, PCollectionView<ActualT>> actualView;
 
     public static <T, ActualT> CreateActual<T, ActualT> from(
         PCollection<T> actual,
         AssertionWindows rewindowActuals,
-        SimpleFunction<Iterable<WindowedValue<T>>, Iterable<T>> extractPane,
+        SimpleFunction<Iterable<ValueInSingleWindow<T>>, Iterable<T>> extractPane,
         PTransform<PCollection<T>, PCollectionView<ActualT>> actualView) {
       return new CreateActual<>(actual, rewindowActuals, extractPane, actualView);
     }
@@ -752,7 +754,7 @@ public class PAssert {
     private CreateActual(
         PCollection<T> actual,
         AssertionWindows rewindowActuals,
-        SimpleFunction<Iterable<WindowedValue<T>>, Iterable<T>> extractPane,
+        SimpleFunction<Iterable<ValueInSingleWindow<T>>, Iterable<T>> extractPane,
         PTransform<PCollection<T>, PCollectionView<ActualT>> actualView) {
       this.actual = actual;
       this.rewindowActuals = rewindowActuals;
@@ -822,7 +824,7 @@ public class PAssert {
    * a single empty iterable, even though in practice most runners will not produce any element.
    */
   private static class GroupGlobally<T>
-      extends PTransform<PCollection<T>, PCollection<Iterable<WindowedValue<T>>>>
+      extends PTransform<PCollection<T>, PCollection<Iterable<ValueInSingleWindow<T>>>>
       implements Serializable {
     private final AssertionWindows rewindowingStrategy;
 
@@ -831,20 +833,20 @@ public class PAssert {
     }
 
     @Override
-    public PCollection<Iterable<WindowedValue<T>>> apply(PCollection<T> input) {
+    public PCollection<Iterable<ValueInSingleWindow<T>>> apply(PCollection<T> input) {
       final int combinedKey = 42;
 
       // Remove the triggering on both
       PTransform<
-              PCollection<KV<Integer, Iterable<WindowedValue<T>>>>,
-              PCollection<KV<Integer, Iterable<WindowedValue<T>>>>>
+              PCollection<KV<Integer, Iterable<ValueInSingleWindow<T>>>>,
+              PCollection<KV<Integer, Iterable<ValueInSingleWindow<T>>>>>
           removeTriggering =
-              Window.<KV<Integer, Iterable<WindowedValue<T>>>>triggering(Never.ever())
+              Window.<KV<Integer, Iterable<ValueInSingleWindow<T>>>>triggering(Never.ever())
                   .discardingFiredPanes()
                   .withAllowedLateness(input.getWindowingStrategy().getAllowedLateness());
       // Group the contents by key. If it is empty, this PCollection will be empty, too.
       // Then key it again with a dummy key.
-      PCollection<KV<Integer, Iterable<WindowedValue<T>>>> groupedContents =
+      PCollection<KV<Integer, Iterable<ValueInSingleWindow<T>>>> groupedContents =
           // TODO: Split the filtering from the rewindowing, and apply filtering before the Gather
           // if the grouping of extra records
           input
@@ -852,45 +854,47 @@ public class PAssert {
               .apply("GatherAllOutputs", GatherAllPanes.<T>globally())
               .apply(
                   "RewindowActuals",
-                  rewindowingStrategy.<Iterable<WindowedValue<T>>>windowActuals())
-              .apply("KeyForDummy", WithKeys.<Integer, Iterable<WindowedValue<T>>>of(combinedKey))
+                  rewindowingStrategy.<Iterable<ValueInSingleWindow<T>>>windowActuals())
+              .apply(
+                  "KeyForDummy",
+                  WithKeys.<Integer, Iterable<ValueInSingleWindow<T>>>of(combinedKey))
               .apply("RemoveActualsTriggering", removeTriggering);
 
       // Create another non-empty PCollection that is keyed with a distinct dummy key
-      PCollection<KV<Integer, Iterable<WindowedValue<T>>>> keyedDummy =
+      PCollection<KV<Integer, Iterable<ValueInSingleWindow<T>>>> keyedDummy =
           input
               .getPipeline()
               .apply(
                   Create.of(
                           KV.of(
                               combinedKey,
-                              (Iterable<WindowedValue<T>>)
-                                  Collections.<WindowedValue<T>>emptyList()))
+                              (Iterable<ValueInSingleWindow<T>>)
+                                  Collections.<ValueInSingleWindow<T>>emptyList()))
                       .withCoder(groupedContents.getCoder()))
               .apply(
                   "WindowIntoDummy",
-                  rewindowingStrategy.<KV<Integer, Iterable<WindowedValue<T>>>>windowDummy())
+                  rewindowingStrategy.<KV<Integer, Iterable<ValueInSingleWindow<T>>>>windowDummy())
               .apply("RemoveDummyTriggering", removeTriggering);
 
       // Flatten them together and group by the combined key to get a single element
-      PCollection<KV<Integer, Iterable<Iterable<WindowedValue<T>>>>> dummyAndContents =
+      PCollection<KV<Integer, Iterable<Iterable<ValueInSingleWindow<T>>>>> dummyAndContents =
           PCollectionList.of(groupedContents)
               .and(keyedDummy)
               .apply(
                   "FlattenDummyAndContents",
-                  Flatten.<KV<Integer, Iterable<WindowedValue<T>>>>pCollections())
+                  Flatten.<KV<Integer, Iterable<ValueInSingleWindow<T>>>>pCollections())
               .apply(
                   "NeverTrigger",
-                  Window.<KV<Integer, Iterable<WindowedValue<T>>>>triggering(Never.ever())
+                  Window.<KV<Integer, Iterable<ValueInSingleWindow<T>>>>triggering(Never.ever())
                       .withAllowedLateness(input.getWindowingStrategy().getAllowedLateness())
                       .discardingFiredPanes())
               .apply(
                   "GroupDummyAndContents",
-                  GroupByKey.<Integer, Iterable<WindowedValue<T>>>create());
+                  GroupByKey.<Integer, Iterable<ValueInSingleWindow<T>>>create());
 
       return dummyAndContents
-          .apply(Values.<Iterable<Iterable<WindowedValue<T>>>>create())
-          .apply(ParDo.of(new ConcatFn<WindowedValue<T>>()));
+          .apply(Values.<Iterable<Iterable<ValueInSingleWindow<T>>>>create())
+          .apply(ParDo.of(new ConcatFn<ValueInSingleWindow<T>>()));
     }
   }
 
@@ -909,12 +913,12 @@ public class PAssert {
       implements Serializable {
     private final SerializableFunction<Iterable<T>, Void> checkerFn;
     private final AssertionWindows rewindowingStrategy;
-    private final SimpleFunction<Iterable<WindowedValue<T>>, Iterable<T>> paneExtractor;
+    private final SimpleFunction<Iterable<ValueInSingleWindow<T>>, Iterable<T>> paneExtractor;
 
     private GroupThenAssert(
         SerializableFunction<Iterable<T>, Void> checkerFn,
         AssertionWindows rewindowingStrategy,
-        SimpleFunction<Iterable<WindowedValue<T>>, Iterable<T>> paneExtractor) {
+        SimpleFunction<Iterable<ValueInSingleWindow<T>>, Iterable<T>> paneExtractor) {
       this.checkerFn = checkerFn;
       this.rewindowingStrategy = rewindowingStrategy;
       this.paneExtractor = paneExtractor;
@@ -940,13 +944,14 @@ public class PAssert {
       extends PTransform<PCollection<Iterable<T>>, PDone> implements Serializable {
     private final SerializableFunction<Iterable<T>, Void> checkerFn;
     private final AssertionWindows rewindowingStrategy;
-    private final SimpleFunction<Iterable<WindowedValue<Iterable<T>>>, Iterable<Iterable<T>>>
+    private final SimpleFunction<Iterable<ValueInSingleWindow<Iterable<T>>>, Iterable<Iterable<T>>>
         paneExtractor;
 
     private GroupThenAssertForSingleton(
         SerializableFunction<Iterable<T>, Void> checkerFn,
         AssertionWindows rewindowingStrategy,
-        SimpleFunction<Iterable<WindowedValue<Iterable<T>>>, Iterable<Iterable<T>>> paneExtractor) {
+        SimpleFunction<Iterable<ValueInSingleWindow<Iterable<T>>>, Iterable<Iterable<T>>>
+            paneExtractor) {
       this.checkerFn = checkerFn;
       this.rewindowingStrategy = rewindowingStrategy;
       this.paneExtractor = paneExtractor;

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d9891234/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/PaneExtractors.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/PaneExtractors.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/PaneExtractors.java
index db72a0c..dd1fac9 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/PaneExtractors.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/PaneExtractors.java
@@ -25,14 +25,13 @@ import org.apache.beam.sdk.transforms.PTransform;
 import org.apache.beam.sdk.transforms.SimpleFunction;
 import org.apache.beam.sdk.transforms.windowing.PaneInfo;
 import org.apache.beam.sdk.transforms.windowing.PaneInfo.Timing;
-import org.apache.beam.sdk.util.WindowedValue;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.TypeDescriptor;
 
 /**
- * {@link PTransform PTransforms} which take an {@link Iterable} of {@link WindowedValue
- * WindowedValues} and outputs an {@link Iterable} of all values in the specified pane, dropping the
- * {@link WindowedValue} metadata.
+ * {@link PTransform PTransforms} which take an {@link Iterable} of {@link ValueInSingleWindow
+ * ValueInSingleWindows} and outputs an {@link Iterable} of all values in the specified pane,
+ * dropping the {@link ValueInSingleWindow} metadata.
  *
  * <p>Although all of the method signatures return SimpleFunction, users should ensure to set the
  * coder of any output {@link PCollection}, as appropriate {@link TypeDescriptor TypeDescriptors}
@@ -42,36 +41,36 @@ final class PaneExtractors {
   private PaneExtractors() {
   }
 
-  static <T> SimpleFunction<Iterable<WindowedValue<T>>, Iterable<T>> onlyPane() {
+  static <T> SimpleFunction<Iterable<ValueInSingleWindow<T>>, Iterable<T>> onlyPane() {
     return new ExtractOnlyPane<>();
   }
 
-  static <T> SimpleFunction<Iterable<WindowedValue<T>>, Iterable<T>> onTimePane() {
+  static <T> SimpleFunction<Iterable<ValueInSingleWindow<T>>, Iterable<T>> onTimePane() {
     return new ExtractOnTimePane<>();
   }
 
-  static <T> SimpleFunction<Iterable<WindowedValue<T>>, Iterable<T>> finalPane() {
+  static <T> SimpleFunction<Iterable<ValueInSingleWindow<T>>, Iterable<T>> finalPane() {
     return new ExtractFinalPane<>();
   }
 
-  static <T> SimpleFunction<Iterable<WindowedValue<T>>, Iterable<T>> nonLatePanes() {
+  static <T> SimpleFunction<Iterable<ValueInSingleWindow<T>>, Iterable<T>> nonLatePanes() {
     return new ExtractNonLatePanes<>();
   }
 
-  static <T> SimpleFunction<Iterable<WindowedValue<T>>, Iterable<T>> earlyPanes() {
+  static <T> SimpleFunction<Iterable<ValueInSingleWindow<T>>, Iterable<T>> earlyPanes() {
     return new ExtractEarlyPanes<>();
   }
 
-  static <T> SimpleFunction<Iterable<WindowedValue<T>>, Iterable<T>> allPanes() {
+  static <T> SimpleFunction<Iterable<ValueInSingleWindow<T>>, Iterable<T>> allPanes() {
     return new ExtractAllPanes<>();
   }
 
   private static class ExtractOnlyPane<T>
-      extends SimpleFunction<Iterable<WindowedValue<T>>, Iterable<T>> {
+      extends SimpleFunction<Iterable<ValueInSingleWindow<T>>, Iterable<T>> {
     @Override
-    public Iterable<T> apply(Iterable<WindowedValue<T>> input) {
+    public Iterable<T> apply(Iterable<ValueInSingleWindow<T>> input) {
       List<T> outputs = new ArrayList<>();
-      for (WindowedValue<T> value : input) {
+      for (ValueInSingleWindow<T> value : input) {
         checkState(value.getPane().isFirst() && value.getPane().isLast(),
             "Expected elements to be produced by a trigger that fires at most once, but got"
                 + "a value in a pane that is %s. Actual Pane Info: %s",
@@ -85,11 +84,11 @@ final class PaneExtractors {
 
 
   private static class ExtractOnTimePane<T>
-      extends SimpleFunction<Iterable<WindowedValue<T>>, Iterable<T>> {
+      extends SimpleFunction<Iterable<ValueInSingleWindow<T>>, Iterable<T>> {
     @Override
-    public Iterable<T> apply(Iterable<WindowedValue<T>> input) {
+    public Iterable<T> apply(Iterable<ValueInSingleWindow<T>> input) {
       List<T> outputs = new ArrayList<>();
-      for (WindowedValue<T> value : input) {
+      for (ValueInSingleWindow<T> value : input) {
         if (value.getPane().getTiming().equals(Timing.ON_TIME)) {
           outputs.add(value.getValue());
         }
@@ -100,11 +99,11 @@ final class PaneExtractors {
 
 
   private static class ExtractFinalPane<T>
-      extends SimpleFunction<Iterable<WindowedValue<T>>, Iterable<T>> {
+      extends SimpleFunction<Iterable<ValueInSingleWindow<T>>, Iterable<T>> {
     @Override
-    public Iterable<T> apply(Iterable<WindowedValue<T>> input) {
+    public Iterable<T> apply(Iterable<ValueInSingleWindow<T>> input) {
       List<T> outputs = new ArrayList<>();
-      for (WindowedValue<T> value : input) {
+      for (ValueInSingleWindow<T> value : input) {
         if (value.getPane().isLast()) {
           outputs.add(value.getValue());
         }
@@ -115,11 +114,11 @@ final class PaneExtractors {
 
 
   private static class ExtractAllPanes<T>
-      extends SimpleFunction<Iterable<WindowedValue<T>>, Iterable<T>> {
+      extends SimpleFunction<Iterable<ValueInSingleWindow<T>>, Iterable<T>> {
     @Override
-    public Iterable<T> apply(Iterable<WindowedValue<T>> input) {
+    public Iterable<T> apply(Iterable<ValueInSingleWindow<T>> input) {
       List<T> outputs = new ArrayList<>();
-      for (WindowedValue<T> value : input) {
+      for (ValueInSingleWindow<T> value : input) {
         outputs.add(value.getValue());
       }
       return outputs;
@@ -128,11 +127,11 @@ final class PaneExtractors {
 
 
   private static class ExtractNonLatePanes<T>
-      extends SimpleFunction<Iterable<WindowedValue<T>>, Iterable<T>> {
+      extends SimpleFunction<Iterable<ValueInSingleWindow<T>>, Iterable<T>> {
     @Override
-    public Iterable<T> apply(Iterable<WindowedValue<T>> input) {
+    public Iterable<T> apply(Iterable<ValueInSingleWindow<T>> input) {
       List<T> outputs = new ArrayList<>();
-      for (WindowedValue<T> value : input) {
+      for (ValueInSingleWindow<T> value : input) {
         if (value.getPane().getTiming() != PaneInfo.Timing.LATE) {
           outputs.add(value.getValue());
         }
@@ -142,11 +141,11 @@ final class PaneExtractors {
   }
 
   private static class ExtractEarlyPanes<T>
-      extends SimpleFunction<Iterable<WindowedValue<T>>, Iterable<T>> {
+      extends SimpleFunction<Iterable<ValueInSingleWindow<T>>, Iterable<T>> {
     @Override
-    public Iterable<T> apply(Iterable<WindowedValue<T>> input) {
+    public Iterable<T> apply(Iterable<ValueInSingleWindow<T>> input) {
       List<T> outputs = new ArrayList<>();
-      for (WindowedValue<T> value : input) {
+      for (ValueInSingleWindow<T> value : input) {
         if (value.getPane().getTiming() == PaneInfo.Timing.EARLY) {
           outputs.add(value.getValue());
         }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d9891234/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/ValueInSingleWindow.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/ValueInSingleWindow.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/ValueInSingleWindow.java
new file mode 100644
index 0000000..9ec030f
--- /dev/null
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/ValueInSingleWindow.java
@@ -0,0 +1,134 @@
+/*
+ * 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.testing;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.auto.value.AutoValue;
+import com.google.common.collect.ImmutableList;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.coders.InstantCoder;
+import org.apache.beam.sdk.coders.StandardCoder;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.PaneInfo;
+import org.apache.beam.sdk.util.PropertyNames;
+import org.joda.time.Instant;
+
+/**
+ * An immutable tuple of value, timestamp, window, and pane.
+ *
+ * @param <T> the type of the value
+ */
+@AutoValue
+public abstract class ValueInSingleWindow<T> {
+  /** Returns the value of this {@code ValueInSingleWindow}. */
+  @Nullable
+  public abstract T getValue();
+
+  /** Returns the timestamp of this {@code ValueInSingleWindow}. */
+  public abstract Instant getTimestamp();
+
+  /** Returns the window of this {@code ValueInSingleWindow}. */
+  public abstract BoundedWindow getWindow();
+
+  /** Returns the pane of this {@code ValueInSingleWindow} in its window. */
+  public abstract PaneInfo getPane();
+
+  public static <T> ValueInSingleWindow<T> of(
+      T value, Instant timestamp, BoundedWindow window, PaneInfo paneInfo) {
+    return new AutoValue_ValueInSingleWindow<>(value, timestamp, window, paneInfo);
+  }
+
+  /** A coder for {@link ValueInSingleWindow}. */
+  public static class Coder<T> extends StandardCoder<ValueInSingleWindow<T>> {
+    private final org.apache.beam.sdk.coders.Coder<T> valueCoder;
+    private final org.apache.beam.sdk.coders.Coder<BoundedWindow> windowCoder;
+
+    public static <T> Coder<T> of(
+        org.apache.beam.sdk.coders.Coder<T> valueCoder,
+        org.apache.beam.sdk.coders.Coder<? extends BoundedWindow> windowCoder) {
+      return new Coder<>(valueCoder, windowCoder);
+    }
+
+    @JsonCreator
+    public static <T> Coder<T> of(
+        @JsonProperty(PropertyNames.COMPONENT_ENCODINGS)
+            List<org.apache.beam.sdk.coders.Coder<?>> components) {
+      checkArgument(components.size() == 2, "Expecting 2 components, got %s", components.size());
+      @SuppressWarnings("unchecked")
+      org.apache.beam.sdk.coders.Coder<T> valueCoder =
+          (org.apache.beam.sdk.coders.Coder<T>) components.get(0);
+      @SuppressWarnings("unchecked")
+      org.apache.beam.sdk.coders.Coder<BoundedWindow> windowCoder =
+          (org.apache.beam.sdk.coders.Coder<BoundedWindow>) components.get(1);
+      return new Coder<>(valueCoder, windowCoder);
+    }
+
+    @SuppressWarnings({"unchecked", "rawtypes"})
+    Coder(
+        org.apache.beam.sdk.coders.Coder<T> valueCoder,
+        org.apache.beam.sdk.coders.Coder<? extends BoundedWindow> windowCoder) {
+      this.valueCoder = valueCoder;
+      this.windowCoder = (org.apache.beam.sdk.coders.Coder) windowCoder;
+    }
+
+    @Override
+    public void encode(ValueInSingleWindow<T> windowedElem, OutputStream outStream, Context context)
+        throws IOException {
+      Context nestedContext = context.nested();
+      valueCoder.encode(windowedElem.getValue(), outStream, nestedContext);
+      InstantCoder.of().encode(windowedElem.getTimestamp(), outStream, nestedContext);
+      windowCoder.encode(windowedElem.getWindow(), outStream, nestedContext);
+      PaneInfo.PaneInfoCoder.INSTANCE.encode(windowedElem.getPane(), outStream, context);
+    }
+
+    @Override
+    public ValueInSingleWindow<T> decode(InputStream inStream, Context context) throws IOException {
+      Context nestedContext = context.nested();
+      T value = valueCoder.decode(inStream, nestedContext);
+      Instant timestamp = InstantCoder.of().decode(inStream, nestedContext);
+      BoundedWindow window = windowCoder.decode(inStream, nestedContext);
+      PaneInfo pane = PaneInfo.PaneInfoCoder.INSTANCE.decode(inStream, nestedContext);
+      return new AutoValue_ValueInSingleWindow<>(value, timestamp, window, pane);
+    }
+
+    @Override
+    public List<? extends org.apache.beam.sdk.coders.Coder<?>> getCoderArguments() {
+      // Coder arguments are coders for the type parameters of the coder - i.e. only T.
+      return ImmutableList.of(valueCoder);
+    }
+
+    @Override
+    public List<? extends org.apache.beam.sdk.coders.Coder<?>> getComponents() {
+      // Coder components are all inner coders that it uses - i.e. both T and BoundedWindow.
+      return ImmutableList.of(valueCoder, windowCoder);
+    }
+
+    @Override
+    public void verifyDeterministic() throws NonDeterministicException {
+      valueCoder.verifyDeterministic();
+      windowCoder.verifyDeterministic();
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d9891234/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java
index 0c6043f..17fa612 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java
@@ -23,7 +23,6 @@ import static com.google.common.base.Preconditions.checkState;
 import com.google.common.base.Function;
 import com.google.common.base.MoreObjects;
 import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Iterables;
 import com.google.common.collect.Lists;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -35,8 +34,10 @@ import java.util.Map;
 import org.apache.beam.sdk.annotations.Experimental;
 import org.apache.beam.sdk.options.PipelineOptions;
 import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.testing.ValueInSingleWindow;
 import org.apache.beam.sdk.transforms.Combine.CombineFn;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
 import org.apache.beam.sdk.transforms.windowing.PaneInfo;
 import org.apache.beam.sdk.util.SerializableUtils;
 import org.apache.beam.sdk.util.TimerInternals;
@@ -353,10 +354,10 @@ public class DoFnTester<InputT, OutputT> implements AutoCloseable {
   public List<TimestampedValue<OutputT>> peekOutputElementsWithTimestamp() {
     // TODO: Should we return an unmodifiable list?
     return Lists.transform(getImmutableOutput(mainOutputTag),
-        new Function<WindowedValue<OutputT>, TimestampedValue<OutputT>>() {
+        new Function<ValueInSingleWindow<OutputT>, TimestampedValue<OutputT>>() {
           @Override
           @SuppressWarnings("unchecked")
-          public TimestampedValue<OutputT> apply(WindowedValue<OutputT> input) {
+          public TimestampedValue<OutputT> apply(ValueInSingleWindow<OutputT> input) {
             return TimestampedValue.of(input.getValue(), input.getTimestamp());
           }
         });
@@ -378,8 +379,8 @@ public class DoFnTester<InputT, OutputT> implements AutoCloseable {
       TupleTag<OutputT> tag,
       BoundedWindow window) {
     ImmutableList.Builder<TimestampedValue<OutputT>> valuesBuilder = ImmutableList.builder();
-    for (WindowedValue<OutputT> value : getImmutableOutput(tag)) {
-      if (value.getWindows().contains(window)) {
+    for (ValueInSingleWindow<OutputT> value : getImmutableOutput(tag)) {
+      if (value.getWindow().equals(window)) {
         valuesBuilder.add(TimestampedValue.of(value.getValue(), value.getTimestamp()));
       }
     }
@@ -434,10 +435,10 @@ public class DoFnTester<InputT, OutputT> implements AutoCloseable {
   public <T> List<T> peekSideOutputElements(TupleTag<T> tag) {
     // TODO: Should we return an unmodifiable list?
     return Lists.transform(getImmutableOutput(tag),
-        new Function<WindowedValue<T>, T>() {
+        new Function<ValueInSingleWindow<T>, T>() {
           @SuppressWarnings("unchecked")
           @Override
-          public T apply(WindowedValue<T> input) {
+          public T apply(ValueInSingleWindow<T> input) {
             return input.getValue();
           }});
   }
@@ -510,16 +511,16 @@ public class DoFnTester<InputT, OutputT> implements AutoCloseable {
     return combiner.extractOutput(accumulator);
   }
 
-  private <T> List<WindowedValue<T>> getImmutableOutput(TupleTag<T> tag) {
+  private <T> List<ValueInSingleWindow<T>> getImmutableOutput(TupleTag<T> tag) {
     @SuppressWarnings({"unchecked", "rawtypes"})
-    List<WindowedValue<T>> elems = (List) outputs.get(tag);
+    List<ValueInSingleWindow<T>> elems = (List) outputs.get(tag);
     return ImmutableList.copyOf(
-        MoreObjects.firstNonNull(elems, Collections.<WindowedValue<T>>emptyList()));
+        MoreObjects.firstNonNull(elems, Collections.<ValueInSingleWindow<T>>emptyList()));
   }
 
   @SuppressWarnings({"unchecked", "rawtypes"})
-  public <T> List<WindowedValue<T>> getMutableOutput(TupleTag<T> tag) {
-    List<WindowedValue<T>> outputList = (List) outputs.get(tag);
+  public <T> List<ValueInSingleWindow<T>> getMutableOutput(TupleTag<T> tag) {
+    List<ValueInSingleWindow<T>> outputList = (List) outputs.get(tag);
     if (outputList == null) {
       outputList = new ArrayList<>();
       outputs.put(tag, (List) outputList);
@@ -612,23 +613,22 @@ public class DoFnTester<InputT, OutputT> implements AutoCloseable {
       sideOutputWithTimestamp(tag, output, BoundedWindow.TIMESTAMP_MIN_VALUE);
     }
 
-    public <T> void noteOutput(TupleTag<T> tag, WindowedValue<T> output) {
+    public <T> void noteOutput(TupleTag<T> tag, ValueInSingleWindow<T> output) {
       getMutableOutput(tag).add(output);
     }
   }
 
   private TestProcessContext createProcessContext(TimestampedValue<InputT> elem) {
-    WindowedValue<InputT> windowedValue = WindowedValue.timestampedValueInGlobalWindow(
-        elem.getValue(), elem.getTimestamp());
-
-    return new TestProcessContext(windowedValue);
+    return new TestProcessContext(
+        ValueInSingleWindow.of(
+            elem.getValue(), elem.getTimestamp(), GlobalWindow.INSTANCE, PaneInfo.NO_FIRING));
   }
 
   private class TestProcessContext extends OldDoFn<InputT, OutputT>.ProcessContext {
     private final TestContext context;
-    private final WindowedValue<InputT> element;
+    private final ValueInSingleWindow<InputT> element;
 
-    private TestProcessContext(WindowedValue<InputT> element) {
+    private TestProcessContext(ValueInSingleWindow<InputT> element) {
       fn.super();
       this.context = createContext(fn);
       this.element = element;
@@ -661,7 +661,7 @@ public class DoFnTester<InputT, OutputT> implements AutoCloseable {
 
     @Override
     public BoundedWindow window() {
-      return Iterables.getOnlyElement(element.getWindows());
+      return element.getWindow();
     }
 
     @Override
@@ -683,7 +683,10 @@ public class DoFnTester<InputT, OutputT> implements AutoCloseable {
             Instant timestamp,
             Collection<? extends BoundedWindow> windows,
             PaneInfo pane) {
-          context.noteOutput(mainOutputTag, WindowedValue.of(output, timestamp, windows, pane));
+          for (BoundedWindow window : windows) {
+            context.noteOutput(
+                mainOutputTag, ValueInSingleWindow.of(output, timestamp, window, pane));
+          }
         }
 
         @Override
@@ -693,7 +696,10 @@ public class DoFnTester<InputT, OutputT> implements AutoCloseable {
             Instant timestamp,
             Collection<? extends BoundedWindow> windows,
             PaneInfo pane) {
-          context.noteOutput(tag, WindowedValue.of(output, timestamp, windows, pane));
+          for (BoundedWindow window : windows) {
+            context.noteOutput(
+                tag, ValueInSingleWindow.of(output, timestamp, window, pane));
+          }
         }
 
         @Override
@@ -703,7 +709,7 @@ public class DoFnTester<InputT, OutputT> implements AutoCloseable {
 
         @Override
         public Collection<? extends BoundedWindow> windows() {
-          return element.getWindows();
+          return Collections.singleton(element.getWindow());
         }
 
         @Override
@@ -742,8 +748,8 @@ public class DoFnTester<InputT, OutputT> implements AutoCloseable {
 
     @Override
     public <T> void sideOutputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp) {
-      context.noteOutput(tag,
-          WindowedValue.of(output, timestamp, element.getWindows(), element.getPane()));
+      context.noteOutput(
+          tag, ValueInSingleWindow.of(output, timestamp, element.getWindow(), element.getPane()));
     }
 
     @Override
@@ -803,7 +809,7 @@ public class DoFnTester<InputT, OutputT> implements AutoCloseable {
   OldDoFn<InputT, OutputT> fn;
 
   /** The outputs from the {@link DoFn} under test. */
-  private Map<TupleTag<?>, List<WindowedValue<?>>> outputs;
+  private Map<TupleTag<?>, List<ValueInSingleWindow<?>>> outputs;
 
   private InMemoryStateInternals<?> stateInternals;
   private InMemoryTimerInternals timerInternals;

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d9891234/sdks/java/core/src/main/java/org/apache/beam/sdk/util/GatherAllPanes.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/GatherAllPanes.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/GatherAllPanes.java
deleted file mode 100644
index 52a2ba8..0000000
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/GatherAllPanes.java
+++ /dev/null
@@ -1,86 +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.util;
-
-import org.apache.beam.sdk.transforms.DoFn;
-import org.apache.beam.sdk.transforms.GroupByKey;
-import org.apache.beam.sdk.transforms.PTransform;
-import org.apache.beam.sdk.transforms.ParDo;
-import org.apache.beam.sdk.transforms.Values;
-import org.apache.beam.sdk.transforms.WithKeys;
-import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
-import org.apache.beam.sdk.transforms.windowing.Never;
-import org.apache.beam.sdk.transforms.windowing.Window;
-import org.apache.beam.sdk.transforms.windowing.WindowFn;
-import org.apache.beam.sdk.values.KV;
-import org.apache.beam.sdk.values.PCollection;
-import org.apache.beam.sdk.values.TypeDescriptor;
-
-/**
- * Gathers all panes of each window into exactly one output.
- *
- * <p>Note that this will delay the output of a window until the garbage collection time (when the
- * watermark passes the end of the window plus allowed lateness) even if the upstream triggers
- * closed the window earlier.
- */
-public class GatherAllPanes<T>
-    extends PTransform<PCollection<T>, PCollection<Iterable<WindowedValue<T>>>> {
-  /**
-   * Gathers all panes of each window into a single output element.
-   *
-   * <p>This will gather all output panes into a single element, which causes them to be colocated
-   * on a single worker. As a result, this is only suitable for {@link PCollection PCollections}
-   * where all of the output elements for each pane fit in memory, such as in tests.
-   */
-  public static <T> GatherAllPanes<T> globally() {
-    return new GatherAllPanes<>();
-  }
-
-  private GatherAllPanes() {}
-
-  @Override
-  public PCollection<Iterable<WindowedValue<T>>> apply(PCollection<T> input) {
-    WindowFn<?, ?> originalWindowFn = input.getWindowingStrategy().getWindowFn();
-
-    return input
-        .apply(ParDo.of(new ReifyTimestampsAndWindowsFn<T>()))
-        .setCoder(
-            WindowedValue.FullWindowedValueCoder.of(
-                input.getCoder(), input.getWindowingStrategy().getWindowFn().windowCoder()))
-        .apply(
-            WithKeys.<Integer, WindowedValue<T>>of(0).withKeyType(new TypeDescriptor<Integer>() {}))
-        .apply(
-            Window.into(
-                    new IdentityWindowFn<KV<Integer, WindowedValue<T>>>(
-                        originalWindowFn.windowCoder()))
-                .triggering(Never.ever())
-                .withAllowedLateness(input.getWindowingStrategy().getAllowedLateness())
-                .discardingFiredPanes())
-        // all values have the same key so they all appear as a single output element
-        .apply(GroupByKey.<Integer, WindowedValue<T>>create())
-        .apply(Values.<Iterable<WindowedValue<T>>>create())
-        .setWindowingStrategyInternal(input.getWindowingStrategy());
-  }
-
-  private static class ReifyTimestampsAndWindowsFn<T> extends DoFn<T, WindowedValue<T>> {
-    @DoFn.ProcessElement
-    public void processElement(ProcessContext c, BoundedWindow window) {
-      c.output(WindowedValue.of(c.element(), c.timestamp(), window, c.pane()));
-    }
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d9891234/sdks/java/core/src/main/java/org/apache/beam/sdk/util/IdentityWindowFn.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/IdentityWindowFn.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/IdentityWindowFn.java
index 8ca1bfd..c02e1f4 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/IdentityWindowFn.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/IdentityWindowFn.java
@@ -45,7 +45,7 @@ import org.joda.time.Instant;
  * <p>This {@link WindowFn} is an internal implementation detail of sdk-provided utilities, and
  * should not be used by {@link Pipeline} writers.
  */
-class IdentityWindowFn<T> extends NonMergingWindowFn<T, BoundedWindow> {
+public class IdentityWindowFn<T> extends NonMergingWindowFn<T, BoundedWindow> {
 
   /**
    * The coder of the type of windows of the input {@link PCollection}. This is not an arbitrary

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d9891234/sdks/java/core/src/test/java/org/apache/beam/sdk/WindowMatchers.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/WindowMatchers.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/WindowMatchers.java
deleted file mode 100644
index 3531a86..0000000
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/WindowMatchers.java
+++ /dev/null
@@ -1,204 +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;
-
-import com.google.common.collect.Lists;
-import java.util.Collection;
-import java.util.Objects;
-import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
-import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
-import org.apache.beam.sdk.transforms.windowing.PaneInfo;
-import org.apache.beam.sdk.util.WindowedValue;
-import org.hamcrest.Description;
-import org.hamcrest.Matcher;
-import org.hamcrest.Matchers;
-import org.hamcrest.TypeSafeMatcher;
-import org.joda.time.Instant;
-
-/**
- * Matchers that are useful for working with Windowing, Timestamps, etc.
- */
-public class WindowMatchers {
-
-  public static <T> Matcher<WindowedValue<? extends T>> isWindowedValue(
-      T value,
-      Instant timestamp,
-      Collection<? extends BoundedWindow> windows,
-      PaneInfo paneInfo) {
-
-    Collection<Matcher<? super BoundedWindow>> windowMatchers =
-        Lists.newArrayListWithCapacity(windows.size());
-    for (BoundedWindow window : windows) {
-      windowMatchers.add(Matchers.equalTo(window));
-    }
-
-    return isWindowedValue(
-        Matchers.equalTo(value),
-        Matchers.equalTo(timestamp),
-        Matchers.containsInAnyOrder(windowMatchers),
-        Matchers.equalTo(paneInfo));
-  }
-
-  public static <T> Matcher<WindowedValue<? extends T>> isWindowedValue(
-      Matcher<? super T> valueMatcher,
-      Matcher<? super Instant> timestampMatcher,
-      Matcher<? super Collection<? extends BoundedWindow>> windowsMatcher,
-      Matcher<? super PaneInfo> paneInfoMatcher) {
-    return new WindowedValueMatcher<>(
-        valueMatcher, timestampMatcher, windowsMatcher, paneInfoMatcher);
-  }
-
-  public static <T> Matcher<WindowedValue<? extends T>> isWindowedValue(
-      Matcher<? super T> valueMatcher,
-      Matcher<? super Instant> timestampMatcher,
-      Matcher<? super Collection<? extends BoundedWindow>> windowsMatcher) {
-    return new WindowedValueMatcher<>(
-        valueMatcher, timestampMatcher, windowsMatcher, Matchers.anything());
-  }
-
-  public static <T> Matcher<WindowedValue<? extends T>> isWindowedValue(
-      Matcher<? super T> valueMatcher, Matcher<? super Instant> timestampMatcher) {
-    return new WindowedValueMatcher<>(
-        valueMatcher, timestampMatcher, Matchers.anything(), Matchers.anything());
-  }
-
-  public static <T> Matcher<WindowedValue<? extends T>> isWindowedValue(
-      Matcher<? super T> valueMatcher) {
-    return new WindowedValueMatcher<>(
-        valueMatcher, Matchers.anything(), Matchers.anything(), Matchers.anything());
-  }
-
-  public static <T> Matcher<WindowedValue<? extends T>> isSingleWindowedValue(
-      T value, long timestamp, long windowStart, long windowEnd) {
-    return WindowMatchers.<T>isSingleWindowedValue(
-        Matchers.equalTo(value), timestamp, windowStart, windowEnd);
-  }
-
-  public static <T> Matcher<WindowedValue<? extends T>> isSingleWindowedValue(
-      T value, Instant timestamp, BoundedWindow window, PaneInfo paneInfo) {
-    return WindowMatchers.<T>isSingleWindowedValue(
-        Matchers.equalTo(value),
-        Matchers.equalTo(timestamp),
-        Matchers.equalTo(window),
-        Matchers.equalTo(paneInfo));
-  }
-
-  public static <T> Matcher<WindowedValue<? extends T>> isSingleWindowedValue(
-      T value, Instant timestamp, BoundedWindow window) {
-    return WindowMatchers.<T>isSingleWindowedValue(
-        Matchers.equalTo(value), Matchers.equalTo(timestamp), Matchers.equalTo(window));
-  }
-
-  public static <T> Matcher<WindowedValue<? extends T>> isSingleWindowedValue(
-      Matcher<T> valueMatcher, long timestamp, long windowStart, long windowEnd) {
-    IntervalWindow intervalWindow =
-        new IntervalWindow(new Instant(windowStart), new Instant(windowEnd));
-    return WindowMatchers.<T>isSingleWindowedValue(
-        valueMatcher,
-        Matchers.describedAs("%0", Matchers.equalTo(new Instant(timestamp)), timestamp),
-        Matchers.<BoundedWindow>equalTo(intervalWindow),
-        Matchers.anything());
-  }
-
-  public static <T> Matcher<WindowedValue<? extends T>> isSingleWindowedValue(
-      Matcher<? super T> valueMatcher,
-      Matcher<? super Instant> timestampMatcher,
-      Matcher<? super BoundedWindow> windowMatcher) {
-    return new WindowedValueMatcher<T>(
-        valueMatcher, timestampMatcher, Matchers.contains(windowMatcher), Matchers.anything());
-  }
-
-  public static <T> Matcher<WindowedValue<? extends T>> isSingleWindowedValue(
-      Matcher<? super T> valueMatcher,
-      Matcher<? super Instant> timestampMatcher,
-      Matcher<? super BoundedWindow> windowMatcher,
-      Matcher<? super PaneInfo> paneInfoMatcher) {
-    return new WindowedValueMatcher<T>(
-        valueMatcher, timestampMatcher, Matchers.contains(windowMatcher), paneInfoMatcher);
-  }
-
-  public static Matcher<IntervalWindow> intervalWindow(long start, long end) {
-    return Matchers.equalTo(new IntervalWindow(new Instant(start), new Instant(end)));
-  }
-
-  public static <T> Matcher<WindowedValue<? extends T>> valueWithPaneInfo(final PaneInfo paneInfo) {
-    return new TypeSafeMatcher<WindowedValue<? extends T>>() {
-      @Override
-      public void describeTo(Description description) {
-        description
-            .appendText("WindowedValue(paneInfo = ").appendValue(paneInfo).appendText(")");
-      }
-
-      @Override
-      protected boolean matchesSafely(WindowedValue<? extends T> item) {
-        return Objects.equals(item.getPane(), paneInfo);
-      }
-
-      @Override
-      protected void describeMismatchSafely(
-          WindowedValue<? extends T> item, Description mismatchDescription) {
-        mismatchDescription.appendValue(item.getPane());
-      }
-    };
-  }
-
-  @SuppressWarnings({"unchecked", "rawtypes"})
-  @SafeVarargs
-  public static final <W extends BoundedWindow> Matcher<Iterable<W>> ofWindows(
-      Matcher<W>... windows) {
-    return (Matcher) Matchers.<W>containsInAnyOrder(windows);
-  }
-
-  private WindowMatchers() {}
-
-  private static class WindowedValueMatcher<T> extends TypeSafeMatcher<WindowedValue<? extends T>> {
-
-    private Matcher<? super T> valueMatcher;
-    private Matcher<? super Instant> timestampMatcher;
-    private Matcher<? super Collection<? extends BoundedWindow>> windowsMatcher;
-    private Matcher<? super PaneInfo> paneInfoMatcher;
-
-    private WindowedValueMatcher(
-        Matcher<? super T> valueMatcher,
-        Matcher<? super Instant> timestampMatcher,
-        Matcher<? super Collection<? extends BoundedWindow>> windowsMatcher,
-        Matcher<? super PaneInfo> paneInfoMatcher) {
-      this.valueMatcher = valueMatcher;
-      this.timestampMatcher = timestampMatcher;
-      this.windowsMatcher = windowsMatcher;
-      this.paneInfoMatcher = paneInfoMatcher;
-    }
-
-    @Override
-    public void describeTo(Description description) {
-      description
-          .appendText("a WindowedValue(").appendValue(valueMatcher)
-          .appendText(", ").appendValue(timestampMatcher)
-          .appendText(", ").appendValue(windowsMatcher)
-          .appendText(", ").appendValue(paneInfoMatcher)
-          .appendText(")");
-    }
-
-    @Override
-    protected boolean matchesSafely(WindowedValue<? extends T> windowedValue) {
-      return valueMatcher.matches(windowedValue.getValue())
-          && timestampMatcher.matches(windowedValue.getTimestamp())
-          && windowsMatcher.matches(windowedValue.getWindows());
-    }
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d9891234/sdks/java/core/src/test/java/org/apache/beam/sdk/WindowMatchersTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/WindowMatchersTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/WindowMatchersTest.java
deleted file mode 100644
index 89637e2..0000000
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/WindowMatchersTest.java
+++ /dev/null
@@ -1,82 +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;
-
-import static org.junit.Assert.assertThat;
-
-import com.google.common.collect.ImmutableList;
-import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
-import org.apache.beam.sdk.transforms.windowing.PaneInfo;
-import org.apache.beam.sdk.util.WindowedValue;
-import org.joda.time.Instant;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.JUnit4;
-
-/**
- * Tests for {@link WindowMatchers}.
- */
-@RunWith(JUnit4.class)
-public class WindowMatchersTest {
-
-  @Test
-  public void testIsWindowedValueExact() {
-    long timestamp = 100;
-    long windowStart = 0;
-    long windowEnd = 200;
-
-    assertThat(
-        WindowedValue.of(
-            "hello",
-            new Instant(timestamp),
-            new IntervalWindow(new Instant(windowStart), new Instant(windowEnd)),
-            PaneInfo.NO_FIRING),
-        WindowMatchers.isWindowedValue(
-            "hello",
-            new Instant(timestamp),
-            ImmutableList.of(new IntervalWindow(new Instant(windowStart), new Instant(windowEnd))),
-            PaneInfo.NO_FIRING));
-  }
-
-  @Test
-  public void testIsWindowedValueReorderedWindows() {
-    long timestamp = 100;
-    long windowStart = 0;
-    long windowEnd = 200;
-    long windowStart2 = 50;
-    long windowEnd2 = 150;
-
-    assertThat(
-        WindowedValue.of(
-            "hello",
-            new Instant(timestamp),
-            ImmutableList.of(
-                new IntervalWindow(new Instant(windowStart), new Instant(windowEnd)),
-                new IntervalWindow(new Instant(windowStart2), new Instant(windowEnd2))),
-            PaneInfo.NO_FIRING),
-        WindowMatchers.isWindowedValue(
-            "hello",
-            new Instant(timestamp),
-            ImmutableList.of(
-                new IntervalWindow(new Instant(windowStart), new Instant(windowEnd)),
-                new IntervalWindow(new Instant(windowStart2), new Instant(windowEnd2))),
-            PaneInfo.NO_FIRING));
-  }
-}
-
-

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d9891234/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/GatherAllPanesTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/GatherAllPanesTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/GatherAllPanesTest.java
new file mode 100644
index 0000000..417147f
--- /dev/null
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/GatherAllPanesTest.java
@@ -0,0 +1,140 @@
+/*
+ * 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.testing;
+
+import static org.junit.Assert.fail;
+
+import com.google.common.collect.Iterables;
+import java.io.Serializable;
+import org.apache.beam.sdk.io.CountingInput;
+import org.apache.beam.sdk.transforms.Flatten;
+import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.transforms.SerializableFunction;
+import org.apache.beam.sdk.transforms.Values;
+import org.apache.beam.sdk.transforms.WithKeys;
+import org.apache.beam.sdk.transforms.WithTimestamps;
+import org.apache.beam.sdk.transforms.windowing.AfterPane;
+import org.apache.beam.sdk.transforms.windowing.AfterWatermark;
+import org.apache.beam.sdk.transforms.windowing.FixedWindows;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionList;
+import org.apache.beam.sdk.values.TypeDescriptor;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link GatherAllPanes}. */
+@RunWith(JUnit4.class)
+public class GatherAllPanesTest implements Serializable {
+  @Test
+  @Category(NeedsRunner.class)
+  public void singlePaneSingleReifiedPane() {
+    TestPipeline p = TestPipeline.create();
+    PCollection<Iterable<ValueInSingleWindow<Iterable<Long>>>> accumulatedPanes =
+        p.apply(CountingInput.upTo(20000))
+            .apply(
+                WithTimestamps.of(
+                    new SerializableFunction<Long, Instant>() {
+                      @Override
+                      public Instant apply(Long input) {
+                        return new Instant(input * 10);
+                      }
+                    }))
+            .apply(
+                Window.<Long>into(FixedWindows.of(Duration.standardMinutes(1)))
+                    .triggering(AfterWatermark.pastEndOfWindow())
+                    .withAllowedLateness(Duration.ZERO)
+                    .discardingFiredPanes())
+            .apply(WithKeys.<Void, Long>of((Void) null).withKeyType(new TypeDescriptor<Void>() {}))
+            .apply(GroupByKey.<Void, Long>create())
+            .apply(Values.<Iterable<Long>>create())
+            .apply(GatherAllPanes.<Iterable<Long>>globally());
+
+    PAssert.that(accumulatedPanes)
+        .satisfies(
+            new SerializableFunction<
+                Iterable<Iterable<ValueInSingleWindow<Iterable<Long>>>>, Void>() {
+              @Override
+              public Void apply(Iterable<Iterable<ValueInSingleWindow<Iterable<Long>>>> input) {
+                for (Iterable<ValueInSingleWindow<Iterable<Long>>> windowedInput : input) {
+                  if (Iterables.size(windowedInput) > 1) {
+                    fail("Expected all windows to have exactly one pane, got " + windowedInput);
+                    return null;
+                  }
+                }
+                return null;
+              }
+            });
+
+    p.run();
+  }
+
+  @Test
+  @Category(NeedsRunner.class)
+  public void multiplePanesMultipleReifiedPane() {
+    TestPipeline p = TestPipeline.create();
+
+    PCollection<Long> someElems = p.apply("someLongs", CountingInput.upTo(20000));
+    PCollection<Long> otherElems = p.apply("otherLongs", CountingInput.upTo(20000));
+    PCollection<Iterable<ValueInSingleWindow<Iterable<Long>>>> accumulatedPanes =
+        PCollectionList.of(someElems)
+            .and(otherElems)
+            .apply(Flatten.<Long>pCollections())
+            .apply(
+                WithTimestamps.of(
+                    new SerializableFunction<Long, Instant>() {
+                      @Override
+                      public Instant apply(Long input) {
+                        return new Instant(input * 10);
+                      }
+                    }))
+            .apply(
+                Window.<Long>into(FixedWindows.of(Duration.standardMinutes(1)))
+                    .triggering(
+                        AfterWatermark.pastEndOfWindow()
+                            .withEarlyFirings(AfterPane.elementCountAtLeast(1)))
+                    .withAllowedLateness(Duration.ZERO)
+                    .discardingFiredPanes())
+            .apply(WithKeys.<Void, Long>of((Void) null).withKeyType(new TypeDescriptor<Void>() {}))
+            .apply(GroupByKey.<Void, Long>create())
+            .apply(Values.<Iterable<Long>>create())
+            .apply(GatherAllPanes.<Iterable<Long>>globally());
+
+    PAssert.that(accumulatedPanes)
+        .satisfies(
+            new SerializableFunction<
+                Iterable<Iterable<ValueInSingleWindow<Iterable<Long>>>>, Void>() {
+              @Override
+              public Void apply(Iterable<Iterable<ValueInSingleWindow<Iterable<Long>>>> input) {
+                for (Iterable<ValueInSingleWindow<Iterable<Long>>> windowedInput : input) {
+                  if (Iterables.size(windowedInput) > 1) {
+                    return null;
+                  }
+                }
+                fail("Expected at least one window to have multiple panes");
+                return null;
+              }
+            });
+
+    p.run();
+  }
+}


Mime
View raw message