public <InputT, OutputT, CollectionT extends PCollection<? extends InputT>>
      DataStream<OutputT> applyBeamPTransform(
          DataStream<InputT> input, PTransform<CollectionT, PCollection<OutputT>> transform) {
    return (DataStream)
        getNonNull(
            applyBeamPTransformInternal(
                ImmutableMap.of("input", input),
                (pipeline, map) -> (CollectionT) getNonNull(map, "input"),
                (output) -> ImmutableMap.of("output", output),
                transform,
                input.getExecutionEnvironment()),
            "output");
  }
@Test
  public void testApplyMultiInputTransform() throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();

    DataStream<String> input1 = env.fromCollection(ImmutableList.of("a", "b", "c"));
    DataStream<String> input2 = env.fromCollection(ImmutableList.of("d", "e", "f"));
    DataStream<String> result =
        new BeamFlinkDataStreamAdapter()
            .applyBeamPTransform(
                ImmutableMap.of("x", input1, "y", input2),
                new PTransform<PCollectionTuple, PCollection<String>>() {
                  @Override
                  public PCollection<String> expand(PCollectionTuple input) {
                    return PCollectionList.of(input.<String>get("x").apply(withPrefix("x")))
                        .and(input.<String>get("y").apply(withPrefix("y")))
                        .apply(Flatten.pCollections());
                  }
                });

    assertThat(
        ImmutableList.copyOf(result.executeAndCollect()),
        containsInAnyOrder("xa", "xb", "xc", "yd", "ye", "yf"));
  }