public static PTransformMatcher flattenWithDuplicateInputs() {
    return new PTransformMatcher() {
      @Override
      public boolean matches(AppliedPTransform<?, ?, ?> application) {
        if (application.getTransform() instanceof Flatten.PCollections) {
          Set<PValue> observed = new HashSet<>();
          for (PValue pvalue : application.getInputs().values()) {
            boolean firstInstance = observed.add(pvalue);
            if (!firstInstance) {
              return true;
            }
          }
        }
        return false;
      }

      @Override
      public String toString() {
        return MoreObjects.toStringHelper("FlattenWithDuplicateInputsMatcher").toString();
      }
    };
  }
@Test
  public void flattenWithDuplicateInputsWithDuplicates() {
    PCollection<Integer> duplicate =
        PCollection.createPrimitiveOutputInternal(
            p, WindowingStrategy.globalDefault(), IsBounded.BOUNDED, VarIntCoder.of());
    AppliedPTransform application =
        AppliedPTransform.of(
            "Flatten",
            ImmutableMap.<TupleTag<?>, PCollection<?>>builder()
                .put(new TupleTag<Integer>(), duplicate)
                .put(new TupleTag<Integer>(), duplicate)
                .build(),
            Collections.singletonMap(
                new TupleTag<Integer>(),
                PCollection.createPrimitiveOutputInternal(
                    p, WindowingStrategy.globalDefault(), IsBounded.BOUNDED, VarIntCoder.of())),
            Flatten.pCollections(),
            ResourceHints.create(),
            p);

    assertThat(PTransformMatchers.flattenWithDuplicateInputs().matches(application), is(true));
  }