Reify { public static <K, V> PTransform<PCollection<KV<K, V>>, PCollection<KV<K, TimestampedValue<V>>>> timestampsInValue() { return new TimestampInValue<>(); } 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 timestampedValuesSucceeds() { 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, TimestampedValue<Integer>>> reified = timestamped.apply(Reify.timestampsInValue()); PAssert.that(reified) .containsInAnyOrder( KV.of("foo", TimestampedValue.of(0, new Instant(0))), KV.of("foo", TimestampedValue.of(1, new Instant(1))), KV.of("bar", TimestampedValue.of(2, new Instant(2))), KV.of("baz", TimestampedValue.of(3, new Instant(3)))); pipeline.run(); }