@Override
  public PTransformReplacement<PCollectionList<T>, PCollection<T>> getReplacementTransform(
      AppliedPTransform<PCollectionList<T>, PCollection<T>, PCollections<T>> transform) {
    checkArgument(
        transform.getInputs().isEmpty(),
        "Unexpected nonempty input %s for %s",
        transform.getInputs(),
        getClass().getSimpleName());
    return PTransformReplacement.of(
        PCollectionList.empty(transform.getPipeline()), new CreateEmptyFromList<T>());
  }
@Test
  public void getInputNonEmptyThrows() {
    PCollectionList<Long> nonEmpty =
        PCollectionList.of(pipeline.apply("unbounded", GenerateSequence.from(0)))
            .and(pipeline.apply("bounded", GenerateSequence.from(0).to(100)));
    thrown.expect(IllegalArgumentException.class);
    thrown.expectMessage(nonEmpty.expand().toString());
    thrown.expectMessage(EmptyFlattenAsCreateFactory.class.getSimpleName());
    factory.getReplacementTransform(
        AppliedPTransform.of(
            "nonEmptyInput",
            PValues.expandInput(nonEmpty),
            Collections.emptyMap(),
            Flatten.pCollections(),
            ResourceHints.create(),
            pipeline));
  }