public static Read<String> readStrings() {
    return Read.newBuilder(
            (PubsubMessage message) -> new String(message.getPayload(), StandardCharsets.UTF_8))
        .setCoder(StringUtf8Coder.of())
        .build();
  }
@Test
  public void testFailedParseWithErrorHandlerConfigured() throws Exception {
    ByteString data = ByteString.copyFrom("Hello, World!".getBytes(StandardCharsets.UTF_8));
    RuntimeException exception = new RuntimeException("Some error message");
    ImmutableList<IncomingMessage> expectedReads =
        ImmutableList.of(
            IncomingMessage.of(
                com.google.pubsub.v1.PubsubMessage.newBuilder().setData(data).build(),
                1234L,
                0,
                UUID.randomUUID().toString(),
                UUID.randomUUID().toString()));
    ImmutableList<OutgoingMessage> expectedWrites = ImmutableList.of();
    clientFactory =
        PubsubTestClient.createFactoryForPullAndPublish(
            SUBSCRIPTION, TOPIC, CLOCK, 60, expectedReads, expectedWrites, ImmutableList.of());

    ErrorHandler<BadRecord, PCollection<Long>> errorHandler =
        pipeline.registerBadRecordErrorHandler(new ErrorSinkTransform());
    PCollection<String> read =
        pipeline.apply(
            PubsubIO.readStrings()
                .fromSubscription(SUBSCRIPTION.getPath())
                .withErrorHandler(errorHandler)
                .withClock(CLOCK)
                .withClientFactory(clientFactory)
                .withCoderAndParseFn(
                    StringUtf8Coder.of(),
                    SimpleFunction.fromSerializableFunctionWithOutputType(
                        message -> {
                          throw exception;
                        },
                        TypeDescriptors.strings())));
    errorHandler.close();

    PAssert.thatSingleton(errorHandler.getOutput()).isEqualTo(1L);
    PAssert.that(read).empty();
    pipeline.run();
  }