public static <T> Read<T> readMessagesWithAttributesWithCoderAndParseFn(
      Coder<T> coder, SimpleFunction<PubsubMessage, T> parseFn) {
    return Read.newBuilder(parseFn).setCoder(coder).setNeedsAttributes(true).build();
  }
@Test
  public void testReadMessagesWithAttributesWithCoderAndParseFn() {
    ImmutableList<IncomingMessage> inputs =
        ImmutableList.of(
            messageWithSuffixAttribute("foo", "-some-suffix"),
            messageWithSuffixAttribute("bar", "-some-other-suffix"));
    clientFactory = PubsubTestClient.createFactoryForPull(CLOCK, SUBSCRIPTION, 60, inputs);

    PCollection<String> read =
        pipeline.apply(
            PubsubIO.readMessagesWithAttributesWithCoderAndParseFn(
                    StringUtf8Coder.of(), new AppendSuffixAttributeToStringPayloadParseFn())
                .fromSubscription(SUBSCRIPTION.getPath())
                .withClock(CLOCK)
                .withClientFactory(clientFactory));

    List<String> outputs = ImmutableList.of("foo-some-suffix", "bar-some-other-suffix");
    PAssert.that(read).containsInAnyOrder(outputs);
    pipeline.run();
  }