public static Write<PubsubMessage> writeMessages() {
    return Write.newBuilder()
        .setTopicProvider(null)
        .setTopicFunction(null)
        .setDynamicDestinations(false)
        .build();
  }
@Test
  public void testWriteMalformedMessagesWithErrorHandler() throws Exception {
    OutgoingMessage msg =
        OutgoingMessage.of(
            com.google.pubsub.v1.PubsubMessage.newBuilder()
                .setData(ByteString.copyFromUtf8("foo"))
                .build(),
            0,
            null,
            "projects/project/topics/topic1");

    try (PubsubTestClientFactory factory =
        PubsubTestClient.createFactoryForPublish(null, ImmutableList.of(msg), ImmutableList.of())) {
      TimestampedValue<PubsubMessage> pubsubMsg =
          TimestampedValue.of(
              new PubsubMessage(
                      msg.getMessage().getData().toByteArray(),
                      Collections.emptyMap(),
                      msg.recordId())
                  .withTopic(msg.topic()),
              Instant.ofEpochMilli(msg.getTimestampMsSinceEpoch()));

      TimestampedValue<PubsubMessage> failingPubsubMsg =
          TimestampedValue.of(
              new PubsubMessage(
                      "foo".getBytes(StandardCharsets.UTF_8),
                      Collections.emptyMap(),
                      msg.recordId())
                  .withTopic("badTopic"),
              Instant.ofEpochMilli(msg.getTimestampMsSinceEpoch()));

      PCollection<PubsubMessage> messages =
          pipeline.apply(
              Create.timestamped(ImmutableList.of(pubsubMsg, failingPubsubMsg))
                  .withCoder(PubsubMessageWithTopicCoder.of()));
      messages.setIsBoundedInternal(PCollection.IsBounded.BOUNDED);
      ErrorHandler<BadRecord, PCollection<Long>> badRecordErrorHandler =
          pipeline.registerBadRecordErrorHandler(new ErrorSinkTransform());
      // The most straightforward method to simulate a bad message is to have a format function that
      // deterministically fails based on some value
      messages.apply(
          PubsubIO.writeMessages()
              .toBuilder()
              .setFormatFn(
                  (ValueInSingleWindow<PubsubMessage> messageAndWindow) -> {
                    if (messageAndWindow.getValue().getTopic().equals("badTopic")) {
                      throw new RuntimeException("expected exception");
                    }
                    return messageAndWindow.getValue();
                  })
              .build()
              .to("projects/project/topics/topic1")
              .withClientFactory(factory)
              .withErrorHandler(badRecordErrorHandler));
      badRecordErrorHandler.close();
      PAssert.thatSingleton(badRecordErrorHandler.getOutput()).isEqualTo(1L);
      pipeline.run();
    }
  }