public static <K, V> Write<K, V> write() {
    return new AutoValue_KafkaIO_Write.Builder<K, V>()
        .setWriteRecordsTransform(writeRecords())
        .build();
  }
@Test
  public void testSinkWithSerializationErrors() throws Exception {
    // Attempt to write 10 elements to Kafka, but they will all fail to serialize, and be sent to
    // the DLQ

    int numElements = 10;

    try (MockProducerWrapper producerWrapper =
        new MockProducerWrapper(new FailingLongSerializer())) {

      ProducerSendCompletionThread completionThread =
          new ProducerSendCompletionThread(producerWrapper.mockProducer).start();

      String topic = "test";

      BadRecordErrorHandler<PCollection<Long>> eh =
          p.registerBadRecordErrorHandler(new ErrorSinkTransform());

      p.apply(mkKafkaReadTransform(numElements, new ValueAsTimestampFn()).withoutMetadata())
          .apply(
              KafkaIO.<Integer, Long>write()
                  .withBootstrapServers("none")
                  .withTopic(topic)
                  .withKeySerializer(IntegerSerializer.class)
                  .withValueSerializer(FailingLongSerializer.class)
                  .withInputTimestamp()
                  .withProducerFactoryFn(new ProducerFactoryFn(producerWrapper.producerKey))
                  .withBadRecordErrorHandler(eh));

      eh.close();

      PAssert.thatSingleton(Objects.requireNonNull(eh.getOutput())).isEqualTo(10L);

      p.run();

      completionThread.shutdown();

      verifyProducerRecords(producerWrapper.mockProducer, topic, 0, false, true);
    }
  }