public Set<PValue> getKeyedPValues() {
    checkState(
        finalized, "can't call getKeyedPValues before a Pipeline has been completely traversed");
    return keyedValues;
  }
@Test
  public void unkeyedInputWithKeyPreserving() {

    PCollection<KV<String, Iterable<WindowedValue<KV<String, Integer>>>>> input =
        p.apply(
            Create.of(
                    KV.of(
                        "hello",
                        (Iterable<WindowedValue<KV<String, Integer>>>)
                            Collections.<WindowedValue<KV<String, Integer>>>emptyList()))
                .withCoder(
                    KvCoder.of(
                        StringUtf8Coder.of(),
                        IterableCoder.of(
                            WindowedValue.getValueOnlyCoder(
                                KvCoder.of(StringUtf8Coder.of(), VarIntCoder.of()))))));

    PCollection<KeyedWorkItem<String, KV<String, Integer>>> unkeyed =
        input
            .apply(ParDo.of(new ParDoMultiOverrideFactory.ToKeyedWorkItem<>()))
            .setCoder(
                KeyedWorkItemCoder.of(
                    StringUtf8Coder.of(),
                    KvCoder.of(StringUtf8Coder.of(), VarIntCoder.of()),
                    GlobalWindow.Coder.INSTANCE));

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