public static <T> PCollections<T> pCollections() {
    return new PCollections<>();
  }
@Test
  @Category(ValidatesRunner.class)
  public void testFlattenWithDifferentInputAndOutputCoders2() {
    // This test exists to prevent a regression in Dataflow. It tests a
    // GroupByKey followed by a Flatten with an SDK-specific output coder.
    PCollection<KV<String, Iterable<String>>> flattenInput =
        p.apply(Create.of(LINES))
            .apply(WithKeys.of("a"))
            .setCoder(KvCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of()))
            .apply(GroupByKey.create());
    PCollection<String> output =
        PCollectionList.of(flattenInput)
            .apply(Flatten.pCollections())
            .setCoder(SerializableCoder.of(new TypeDescriptor<KV<String, Iterable<String>>>() {}))
            .apply(Values.create())
            .setCoder(IterableCoder.of(StringUtf8Coder.of()))
            .apply(
                FlatMapElements.into(TypeDescriptors.strings())
                    .via((Iterable<String> values) -> values));
    PAssert.that(output).containsInAnyOrder(LINES);
    p.run();
  }