public static <K, V>
      PTransform<PCollection<KV<K, V>>, PCollection<KV<K, ValueInSingleWindow<V>>>>
          windowsInValue() {
    return new WindowInValue<>();
  }
@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();
  }