public static <K, V> Reshuffle<K, V> of() {
    return new Reshuffle<>();
  }
@Test
  @Category(ValidatesRunner.class)
  public void testReshufflePreservesTimestamps() {
    PCollection<KV<String, TimestampedValue<String>>> input =
        pipeline
            .apply(
                Create.timestamped(
                        TimestampedValue.of("foo", BoundedWindow.TIMESTAMP_MIN_VALUE),
                        TimestampedValue.of("foo", new Instant(0)),
                        TimestampedValue.of("bar", new Instant(33)),
                        TimestampedValue.of("bar", GlobalWindow.INSTANCE.maxTimestamp()))
                    .withCoder(StringUtf8Coder.of()))
            .apply(
                WithKeys.<String, String>of(input12 -> input12)
                    .withKeyType(TypeDescriptors.strings()))
            .apply("ReifyOriginalTimestamps", Reify.timestampsInValue());

    // The outer TimestampedValue is the reified timestamp post-reshuffle. The inner
    // TimestampedValue is the pre-reshuffle timestamp.
    PCollection<TimestampedValue<TimestampedValue<String>>> output =
        input
            .apply(Reshuffle.of())
            .apply("ReifyReshuffledTimestamps", Reify.timestampsInValue())
            .apply(Values.create());

    PAssert.that(output)
        .satisfies(
            input1 -> {
              for (TimestampedValue<TimestampedValue<String>> elem : input1) {
                Instant originalTimestamp = elem.getValue().getTimestamp();
                Instant afterReshuffleTimestamp = elem.getTimestamp();
                assertThat(
                    "Reshuffle must preserve element timestamps",
                    afterReshuffleTimestamp,
                    equalTo(originalTimestamp));
              }
              return null;
            });

    pipeline.run();
  }