Reify { public static <K, V> PTransform<PCollection<KV<K, V>>, PCollection<KV<K, ValueInSingleWindow<V>>>> windowsInValue() { return new WindowInValue<>(); } private  Reify(); static PTransform<PCollection<T>, PCollection<TimestampedValue<T>>> timestamps(); static PTransform<PCollection<KV<K, V>>, PCollection<KV<K, TimestampedValue<V>>>> timestampsInValue(); static PTransform<PCollection<T>, PCollection<ValueInSingleWindow<T>>> windows(); static PTransform<PCollection<KV<K, V>>, PCollection<KV<K, ValueInSingleWindow<V>>>> windowsInValue(); static PTransform<PCollection<KV<K, TimestampedValue<V>>>, PCollection<KV<K, V>>> extractTimestampsFromValues(); static PTransform<PCollection<K>, PCollection<KV<K, V>>> viewAsValues(
      PCollectionView<V> view, Coder<V> coder); static PTransform<PBegin, PCollection<V>> viewInGlobalWindow(
      PCollectionView<V> view, Coder<V> coder);  }
@Test @Category(NeedsRunner.class) public void windowsInValueSucceeds() { PCollection<KV<String, Integer>> timestamped = pipeline .apply(Create.of(KV.of("foo", 0), KV.of("foo", 1), KV.of("bar", 2), KV.of("baz", 3))) .apply(TIMESTAMP_FROM_V); PCollection<KV<String, ValueInSingleWindow<Integer>>> reified = timestamped.apply(Reify.windowsInValue()); PAssert.that(reified) .containsInAnyOrder( KV.of( "foo", ValueInSingleWindow.of( 0, new Instant(0), GlobalWindow.INSTANCE, PaneInfo.NO_FIRING)), KV.of( "foo", ValueInSingleWindow.of( 1, new Instant(1), GlobalWindow.INSTANCE, PaneInfo.NO_FIRING)), KV.of( "bar", ValueInSingleWindow.of( 2, new Instant(2), GlobalWindow.INSTANCE, PaneInfo.NO_FIRING)), KV.of( "baz", ValueInSingleWindow.of( 3, new Instant(3), GlobalWindow.INSTANCE, PaneInfo.NO_FIRING))); pipeline.run(); }