public Set<PValue> getKeyedPValues() {
    checkState(
        finalized, "can't call getKeyedPValues before a Pipeline has been completely traversed");
    return keyedValues;
  }
@Test
  public void groupByKeyProducesKeyedOutput() {
    PCollection<KV<String, Iterable<Integer>>> keyed =
        p.apply(Create.of(KV.of("foo", 3)))
            .apply(new DirectGroupByKeyOnly<>())
            .apply(
                new DirectGroupAlsoByWindow<>(
                    WindowingStrategy.globalDefault(), WindowingStrategy.globalDefault()));

    p.traverseTopologically(visitor);
    assertThat(visitor.getKeyedPValues(), hasItem(keyed));
  }