public static <T> RedistributeArbitrarily<T> arbitrarily() {
    return new RedistributeArbitrarily<>(null, false);
  }
@Test
  @Category(ValidatesRunner.class)
  public void testRedistributePreservesMetadata() {
    PCollection<KV<String, ValueInSingleWindow<String>>> input =
        pipeline
            .apply(
                Create.windowedValues(
                        WindowedValue.of(
                            "foo",
                            BoundedWindow.TIMESTAMP_MIN_VALUE,
                            GlobalWindow.INSTANCE,
                            PaneInfo.NO_FIRING),
                        WindowedValue.of(
                            "foo",
                            new Instant(0),
                            GlobalWindow.INSTANCE,
                            PaneInfo.ON_TIME_AND_ONLY_FIRING),
                        WindowedValue.of(
                            "bar",
                            new Instant(33),
                            GlobalWindow.INSTANCE,
                            PaneInfo.createPane(false, false, PaneInfo.Timing.LATE, 1, 1)),
                        WindowedValue.of(
                            "bar",
                            GlobalWindow.INSTANCE.maxTimestamp(),
                            GlobalWindow.INSTANCE,
                            PaneInfo.NO_FIRING))
                    .withCoder(StringUtf8Coder.of())
                    .withWindowCoder(GlobalWindow.Coder.INSTANCE))
            .apply(WithKeys.<String, String>of(v -> v).withKeyType(TypeDescriptors.strings()))
            .apply("ReifyOriginalMetadata", Reify.windowsInValue());

    // The outer WindowedValue is the reified metadata post-reshuffle. The inner
    // WindowedValue is the pre-reshuffle metadata.
    PCollection<ValueInSingleWindow<ValueInSingleWindow<String>>> output =
        input
            .apply(Redistribute.arbitrarily())
            .apply("ReifyRedistributedMetadata", Reify.windowsInValue())
            .apply(Values.create());

    PAssert.that(output)
        .satisfies(
            input1 -> {
              for (ValueInSingleWindow<ValueInSingleWindow<String>> elem : input1) {
                Instant originalTimestamp = elem.getValue().getTimestamp();
                Instant afterRedistributeTimestamp = elem.getTimestamp();
                assertThat(
                    "Redistribute did not preserve element timestamp for " + elem,
                    afterRedistributeTimestamp,
                    equalTo(originalTimestamp));

                PaneInfo originalPaneInfo = elem.getValue().getPane();
                PaneInfo afterRedistributePaneInfo = elem.getPane();
                assertThat(
                    "Redistribute did not preserve pane info for " + elem,
                    afterRedistributePaneInfo,
                    equalTo(originalPaneInfo));

                BoundedWindow originalWindow = elem.getValue().getWindow();
                BoundedWindow afterRedistributeWindow = elem.getWindow();
                assertThat(
                    "Redistribute did not preserve window for " + elem,
                    afterRedistributeWindow,
                    equalTo(originalWindow));
              }
              return null;
            });

    pipeline.run();
  }